lib

Documentation

Index

• func BuildJobExecutionLockKey(jobExecutionID int64) string
• func BuildJobRequestLockKey(jobExecutionID int64) string
• type DAG
  • func NewDAG[T Node](nodes ...T) (*DAG[T], error)
  • func (d *DAG[T]) DeepPost(node string) ([]T, error)
  • func (d *DAG[T]) DeepPre(node string) ([]T, error)
• type Locker
• type Node
• type ProtobufTypeExtension
  • func (e *ProtobufTypeExtension) CreateDecoder(typ reflect2.Type) jsoniter.ValDecoder
  • func (e *ProtobufTypeExtension) CreateEncoder(typ reflect2.Type) jsoniter.ValEncoder
• type RingBuffer
  • func NewRingBuffer(size int) *RingBuffer
  • func (r *RingBuffer) Close() error
  • func (r *RingBuffer) Read(p []byte) (n int, err error)
  • func (r *RingBuffer) Write(p []byte) (n int, err error)

Functions

func BuildJobExecutionLockKey

func BuildJobExecutionLockKey(jobExecutionID int64) string

BuildJobExecutionLockKey builds a lock key for a job execution. The key is used to lock a job execution to prevent concurrent update job execution.

func BuildJobRequestLockKey

func BuildJobRequestLockKey(jobExecutionID int64) string

BuildJobRequestLockKey builds a lock key for a job request. The key is used to lock a job request to prevent multiple requests for the same job execution. Server keeps the lock for a period, and waits the agent to pick up the job execution. The job request process is a 2 phase commit process. Phase 1 is to lock the job request, and phase 2 is to really get the job execution.

Types

type DAG

type DAG[T Node] struct {
	// contains filtered or unexported fields
}

DAG uses topological sort to build a directed acyclic graph.

func NewDAG

func NewDAG[T Node](nodes ...T) (*DAG[T], error)

NewDAG creates a new DAG.

func (*DAG[T]) DeepPost

func (d *DAG[T]) DeepPost(node string) ([]T, error)

DeepPost returns the deep posting nodes (directly posting and indirectly posting) of the given node.

func (*DAG[T]) DeepPre

func (d *DAG[T]) DeepPre(node string) ([]T, error)

DeepPre returns the deep preceding nodes (directly preceding and indirectly preceding) of the given node.

type Locker

type Locker interface {
	// Lock tries to lock a key with the given value for a period.
	// It returns whether locking was successful and whether an error occurred.
	Lock(ctx context.Context, key, value string, expiresIn time.Duration) (bool, error)
	// Unlock tries to unlock a key. It only succeeds if the value matches.
	Unlock(ctx context.Context, key string) (bool, error)
}

type Node

type Node interface {
	ID() string
	Depends() []string
}

type ProtobufTypeExtension

type ProtobufTypeExtension struct {
	jsoniter.DummyExtension
}

func (*ProtobufTypeExtension) CreateDecoder

func (e *ProtobufTypeExtension) CreateDecoder(typ reflect2.Type) jsoniter.ValDecoder

func (*ProtobufTypeExtension) CreateEncoder

func (e *ProtobufTypeExtension) CreateEncoder(typ reflect2.Type) jsoniter.ValEncoder

type RingBuffer

type RingBuffer struct {
	// contains filtered or unexported fields
}

RingBuffer is a ring buffer with a Close() method.

func NewRingBuffer

func NewRingBuffer(size int) *RingBuffer

NewRingBuffer creates a new ring buffer with given size.

func (*RingBuffer) Close

func (r *RingBuffer) Close() error

Close closes the ring buffer.

func (*RingBuffer) Read

func (r *RingBuffer) Read(p []byte) (n int, err error)

Read reads data from the ring buffer.

func (*RingBuffer) Write

func (r *RingBuffer) Write(p []byte) (n int, err error)

Write writes data to the ring buffer.