pgnotifier

Documentation

Index

• Constants
• func CheckForPGNotifierEnvVars() bool
• func GetCurrentSQLTimestamp() string
• func GetPGNotifierConnectionString() string
• func Init()
• type Claim
• type ClaimResponse
• type JobPayload
• type JobsResponse
• type MessagePayload
• type PGNotifier
  • func New(workspaceIdentifier, fallbackConnectionInfo string) (notifier PGNotifier, err error)
  • func (notifier PGNotifier) ClearJobs(ctx context.Context) (err error)
  • func (notifier PGNotifier) GetDBHandle() *sql.DB
  • func (notifier *PGNotifier) Publish(payload MessagePayload, schema *whUtils.Schema, priority int) (ch chan []Response, err error)
  • func (notifier *PGNotifier) RunMaintenanceWorker(ctx context.Context) error
  • func (notifier *PGNotifier) Subscribe(ctx context.Context, workerId string, jobsBufferSize int) chan Claim
  • func (notifier *PGNotifier) UpdateClaimedEvent(claim *Claim, response *ClaimResponse)
• type Response

Constants

const (
	WaitingState   = "waiting"
	ExecutingState = "executing"
	SucceededState = "succeeded"
	FailedState    = "failed"
	AbortedState   = "aborted"
)

const (
	AsyncJobType  = "async_job"
	UploadJobType = "upload"
)

const MaxTrackAsyncBatchRetries = 5

Functions

func CheckForPGNotifierEnvVars

func CheckForPGNotifierEnvVars() bool

CheckForPGNotifierEnvVars Checks if all the required Env Variables for PG Notifier are present

func GetCurrentSQLTimestamp

func GetCurrentSQLTimestamp() string

GetCurrentSQLTimestamp to get sql complaint current datetime string

func GetPGNotifierConnectionString

func GetPGNotifierConnectionString() string

GetPGNotifierConnectionString Returns PG Notifier DB Connection Configuration

func Init

func Init()

Types

type Claim

type Claim struct {
	ID        int64
	BatchID   string
	Status    string
	Workspace string
	Payload   json.RawMessage
	Attempt   int
	JobType   string
}

type ClaimResponse

type ClaimResponse struct {
	Payload json.RawMessage
	Err     error
}

type JobPayload

type JobPayload json.RawMessage

type JobsResponse

type JobsResponse struct {
	Status    string
	Output    json.RawMessage
	Error     string
	JobType   string
	JobRunID  string
	TaskRunID string
}

type MessagePayload

type MessagePayload struct {
	Jobs    []JobPayload
	JobType string
}

type PGNotifier

type PGNotifier struct {
	URI string
	// contains filtered or unexported fields
}

func New

func New(workspaceIdentifier, fallbackConnectionInfo string) (notifier PGNotifier, err error)

New Given default connection info return pg notifier object from it

func (PGNotifier) ClearJobs

func (notifier PGNotifier) ClearJobs(ctx context.Context) (err error)

func (PGNotifier) GetDBHandle

func (notifier PGNotifier) GetDBHandle() *sql.DB

func (*PGNotifier) Publish

func (notifier *PGNotifier) Publish(payload MessagePayload, schema *whUtils.Schema, priority int) (ch chan []Response, err error)

func (*PGNotifier) RunMaintenanceWorker

func (notifier *PGNotifier) RunMaintenanceWorker(ctx context.Context) error

RunMaintenanceWorker (blocking - to be called from go routine) re-triggers zombie jobs which were left behind by dead workers in executing state

func (*PGNotifier) Subscribe

func (notifier *PGNotifier) Subscribe(ctx context.Context, workerId string, jobsBufferSize int) chan Claim

func (*PGNotifier) UpdateClaimedEvent

func (notifier *PGNotifier) UpdateClaimedEvent(claim *Claim, response *ClaimResponse)

type Response

type Response struct {
	JobID   int64
	Status  string
	Output  json.RawMessage
	Error   string
	JobType string
}