nq

package module

Go to main page

Versions in this module

v0

v0.3.0

Aug 30, 2022

Changes in this version

+ type Inspector struct

+ func NewInspector(broker *NatsBroker) *Inspector

+ func (i *Inspector) AddAnother(queue string, sendUpdatesTo chan string) error

+ func (i *Inspector) Queues()

+ func (i *Inspector) Servers()

+ type ListenUpdates struct

type PublishClient

+ func (p *PublishClient) GetUpdates(taskID string) (chan *TaskMessage, error)

type ResultHandlerIFACE

+ Watch func(id string) (chan *TaskMessage, error)

type ResultHandlerNats

+ func (rn *ResultHandlerNats) Watch(id string) (chan *TaskMessage, error)

v0.2.1

Aug 23, 2022

v0.2.0

Aug 23, 2022

v0.1.0

Aug 23, 2022

Changes in this version

+ const Cancelled

+ const Completed

+ const Deleted

+ const Failed

+ const Pending

+ const Processing

+ var ErrCannotCancelDeletedTask = errors.New("deleted task cannot be cancelled")

+ var ErrFailedToConnect = errors.New("failed to connect to nats client")

+ var ErrInvalidTaskPayload = errors.New("invalid task payload")

+ var ErrNonCancellableState = errors.New("cannot cancel task, in uncancellable state")

+ var ErrQueueNotFound = errors.New("nq: queue not found")

+ var ErrServerClosed = errors.New("nq: Server closed")

+ var ErrServerNameEmpty = errors.New("server name cannot be empty")

+ var ErrStreamNotCreated = errors.New("nq: stream not created")

+ var ErrTaskIDEmpty = errors.New("nq: task id cannot be empty")

+ var ErrTaskNotFound = errors.New("task not found")

+ func CancelStreamNameToStreamName(stream, subject string) string

+ func EncodeTMToJSON(t *TaskMessage) ([]byte, error)

+ func GenerateServerName() string

+ func NoAuthentcation() noAuthentication

+ func ShutdownOnNatsDisconnect() shutdownOnNatsDisconnect

+ func StreamNameToCancelStreamName(subject string) string

+ func StreamNameToDurableStreamName(srvName, stream string) string

+ func TokenAuthentication(username, password string) tokenAuthentication

+ func UserPassAuthentcation(username, password string) uPassAuthentication

+ type CancelPayload string

+ type CancellationStore struct

+ func NewCancelations() CancellationStore

+ func (c *CancellationStore) Add(id string, fn context.CancelFunc)

+ func (c *CancellationStore) Delete(id string)

+ func (c *CancellationStore) Get(id string) (fn context.CancelFunc, ok bool)

+ type ClientConnectionOption interface

+ String func() string

+ Type func() ClientOptionType

+ Value func() interface{}

+ type ClientOption struct

+ AuthenticationObject interface{}

+ AuthenticationType ClientOptionType

+ NatsOption []nats.Option

+ ShutdownOnNatsDisconnect bool

+ Timeout time.Duration

+ type ClientOptionType int

+ const NoAuthenticationOpt

+ const ShutdownOnNatsDisconnectOpt

+ const TokenAuthenticationOpt

+ const UPassAuthenticationOpt

+ type Config struct

+ Concurrency int

+ IsFailureFn func(error) bool

+ LogLevel LogLevel

+ Logger ilog.Base

+ ServerName string

+ ShutdownTimeout time.Duration

+ type LogLevel int32

+ const DebugLevel

+ const ErrorLevel

+ const FatalLevel

+ const InfoLevel

+ const WarnLevel

+ type NatsBroker struct

+ func NewNatsBroker(conf NatsClientOpt, opt ClientOption, natsConnectionClosed chan struct{}, ...) (*NatsBroker, error)

+ func (n *NatsBroker) AddStream(conf nats.StreamConfig) error

+ func (n *NatsBroker) Cancel(subject string, id string) (*TaskMessage, error)

+ func (n *NatsBroker) Close() error

+ func (n *NatsBroker) ConnectoQueue(q *Queue) error

+ func (n *NatsBroker) DeleteStream(name string) error

+ func (n *NatsBroker) Ping() error

+ func (n *NatsBroker) Publish(subject string, payload []byte) (*nats.PubAck, error)

+ func (n *NatsBroker) PublishWithMeta(msg *TaskMessage) (*TaskMessage, error)

+ func (n *NatsBroker) Stats(q *Queue) error

+ func (n *NatsBroker) Submit(subject string, payload TaskPayload) (*TaskMessage, error)

+ type NatsClientOpt struct

+ Addr string

+ DBName string

+ MaxReconnects int

+ ReconnectWait time.Duration

+ type PackagePubAck struct

+ ID string

+ type ProcessingFunc func(context.Context, *TaskPayload) error

+ type PublishClient struct

+ func NewPublishClient(config NatsClientOpt, opts ...ClientConnectionOption) *PublishClient

+ func (p *PublishClient) Cancel(id string) error

+ func (p *PublishClient) CancelInQueue(id string, qname string) error

+ func (p *PublishClient) Close() error

+ func (p *PublishClient) DeleteQueue(qname string)

+ func (p *PublishClient) Enqueue(task *Task, opts ...TaskOption) (*TaskMessage, error)

+ func (p *PublishClient) Fetch(id string) (*TaskMessage, error)

+ func (p *PublishClient) PublishToSubject(task *Task, opts ...TaskOption) (*TaskMessage, error)

+ func (p *PublishClient) Stats(queue string) error

+ type PullAction struct

+ Fn ProcessingFunc

+ Q *Queue

+ Subscription *nats.Subscription

+ type PullStore struct

+ func NewPullStore() PullStore

+ type Queue struct

+ func NewQueue(name string) *Queue

+ func (q *Queue) DurableStream(prefix string) string

+ type ResultHandlerIFACE interface

+ Get func(id string) (*TaskMessage, error)

+ GetAllKeys func(id string, data []byte) ([]string, error)

+ Set func(id string, data []byte) error

+ type ResultHandlerNats struct

+ func NewResultHandlerNats(name string, js nats.JetStreamContext) *ResultHandlerNats

+ func (rn *ResultHandlerNats) Get(id string) (*TaskMessage, error)

+ func (rn *ResultHandlerNats) GetAllKeys(id string, data []byte) ([]string, error)

+ func (rn *ResultHandlerNats) Set(id string, data []byte) error

+ type Server struct

+ func NewServer(natsConfig NatsClientOpt, servCfg Config, opts ...ClientConnectionOption) *Server

+ func (srv *Server) Register(qname string, fn ProcessingFunc)

+ func (srv *Server) Run() error

+ func (srv *Server) Shutdown()

+ func (srv *Server) Start() error

+ func (srv *Server) Stop()

+ type Task struct

+ func NewTask(queue string, payload []byte, opts ...TaskOption) *Task

+ type TaskCancellationMessage struct

+ ID string

+ StreamName string

+ type TaskMessage struct

+ CompletedAt int64

+ CurrentRetry int

+ Deadline int64

+ ID string

+ MaxRetry int

+ Payload []byte

+ Queue string

+ Sequence uint64

+ Status int

+ StreamName string

+ Timeout int64

+ func DecodeTMFromJSON(data []byte) (*TaskMessage, error)

+ func (msg *TaskMessage) GetStatus() string

+ type TaskOption interface

+ String func() string

+ Type func() TaskOptionType

+ Value func() interface{}

+ func Deadline(t time.Time) TaskOption

+ func Retry(n int) TaskOption

+ func TaskID(id string) TaskOption

+ func Timeout(d time.Duration) TaskOption

+ type TaskOptionType int

+ const DeadlineOpt

+ const MaxRetryOpt

+ const TaskIDOpt

+ const TimeoutOpt

+ type TaskPayload struct

+ ID string

+ Payload []byte

v0.1.0-beta.4

Aug 21, 2022

retracted

v0.1.0-beta.3

Aug 15, 2022

retracted

v0.1.0-beta.2

Aug 14, 2022

retracted

v0.1.0-beta.1

Aug 6, 2022

retracted

v0.1.0-beta.0

Aug 6, 2022

retracted

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL