pkg

Documentation

Index

• Constants
• type ChannelHandlers
• type DelayedConsumer
  • func NewDelayedConsumerSet(consumeSubject string, publishSubject string, count int, ...) []*DelayedConsumer
  • func (dc *DelayedConsumer) Start() error
  • func (dc *DelayedConsumer) Stop() error
• type Event
  • func (e Event) String() string
• type EventHandlerCallback
• type EventOctopus
  • func EventOctopusInstance() *EventOctopus
  • func (octopus *EventOctopus) Configure() error
  • func (octopus *EventOctopus) Diagnostics() []core.DiagnosticResult
  • func (octopus *EventOctopus) EventPublisher(clientID string) (IEventPublisher, error)
  • func (octopus *EventOctopus) GetEvent(uuid string) (*Event, error)
  • func (octopus *EventOctopus) GetEventByExternalID(externalID string) (*Event, error)
  • func (octopus *EventOctopus) List() (*[]Event, error)
  • func (octopus *EventOctopus) RunMigrations(db *sql.DB) error
  • func (octopus *EventOctopus) SaveOrUpdateEvent(event Event) error
  • func (octopus *EventOctopus) Shutdown() error
  • func (octopus *EventOctopus) Start() error
  • func (octopus *EventOctopus) Subscribe(service, subject string, handlers map[string]EventHandlerCallback) error
  • func (octopus *EventOctopus) Unsubscribe(service, subject string) error
• type EventOctopusClient
• type EventOctopusConfig
  • func (c EventOctopusConfig) GetMode() string
• type EventPublisher
  • func (p EventPublisher) Publish(subject string, event Event) error
• type IEventPublisher

Constants

const ChannelConsentErrored = "consentRequestErrored"

ChannelConsentErrored can be used to broadcast events which are errored and can not be retried

const ChannelConsentRequest = "consentRequest"

ChannelConsentRequest is the default channel to broadcast events

const ChannelConsentRetry = "consentRequestRetry"

ChannelConsentRetry can be used to broadcast events which are errored and should be retried

const ClientID = "event-store"

ClientID is the Nats client ID

const ConfigAutoRecover = "autoRecover"

ConfigAutoRecover is the config name for republishing unfinished events at startup

const ConfigConnectionStringDefault = "file::memory:?cache=shared"

ConfigConnectionStringDefault defines the default sqlite connection string

const ConfigConnectionstring = "connectionstring"

ConfigConnectionstring defines the string for the flagset

const ConfigIncrementalBackoff = "incrementalBackoff"

ConfigIncrementalBackoff is the name of the config used to determine the incremental backoff

const ConfigIncrementalBackoffDefault = 8

ConfigIncrementalBackoffDefault is the default setting for the incremental backoff of retrying events

const ConfigMaxRetryCount = "maxRetryCount"

ConfigMaxRetryCount is the config name for the number of retries for an event

const ConfigMaxRetryCountDefault = 5

ConfigMaxRetryCountDefault is the default setting for the number of retries

const ConfigNatsPort = "natsPort"

ConfigNatsPort defines the string for the flagset

const ConfigNatsPortDefault = 4222

ConfigNatsPortDefault defines the default nats port

const ConfigPurgeCompleted = "purgeCompleted"

ConfigPurgeCompleted is the config name for enabling purging completed events

const ConfigRetryInterval = "retryInterval"

ConfigRetryInterval defines the string for the flagset

const ConfigRetryIntervalDefault = 60

ConfigRetryIntervalDefault defines the default for the nats retryInterval

const EventAllSignaturesPresent = "all signatures present"

EventAllSignaturesPresent is emitted when all nodes have signed the consentRecord

const EventAttachmentSigned = "attachment signed"

EventAttachmentSigned indicates one of the signatures has been signed

const EventClosed = "closed"

EventErrored indicates the branch was closed

const EventCompleted = "completed"

EventCompleted indicates the flow has been completed 🎉

const EventConsentDistributed = "consent distributed"

EventConsentDistributed indicates the consent request has been distributed among all nodes

const EventConsentRequestAcked = "consentRequest acked"

EventConsentRequestAcked indicates the consentRequest has approved by the vendor system

const EventConsentRequestConstructed = "consentRequest constructed"

EventConsentRequestConstructed is the event emitted directly after consent request creation to start the flow

const EventConsentRequestFlowErrored = "consentRequest flow errored"

EventConsentRequestFlowErrored indicates something went wrong 😔

const EventConsentRequestFlowSuccess = "consentRequest flow success"

EventConsentRequestFlowSuccess is used when the Corda flow has been executed successfully

const EventConsentRequestInFlight = "consentRequest in flight"

EventConsentRequestInFlight is used to indicate the node is waiting for corda to come back with more information

const EventConsentRequestNacked = "consentRequest nacked"

EventConsentRequestNacked indicates the consentRequest has been denied by the vendor system

const EventConsentRequestValid = "consentRequest valid"

EventConsentRequestValid indicates the consentRequest is technically a valid request

const EventDistributedConsentRequestReceived = "distributed ConsentRequest received"

EventDistributedConsentRequestReceived is broadcasted by the consent-bridge when a request has been received

const EventErrored = "error"

EventErrored indicates the flow has errored

const EventInFinalFlight = "consentRequest in flight for final state"

EventInFinalFlight indicates the consentRequest is in flight for final storage

const Name = "Events octopus"

Name is the name of this module

Types

type ChannelHandlers

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

ChannelHandlers store all the handlers for a specific channel subscription

type DelayedConsumer

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

DelayedConsumer holds info for creating a subscription on Nats for consuming events and re-publishing them with a certain delay

func NewDelayedConsumerSet

func NewDelayedConsumerSet(consumeSubject string, publishSubject string, count int, interval time.Duration, exponent int, conn stan.Conn) []*DelayedConsumer

NewDelayedConsumerSet creates a set of DelayedConsumer where each successive poller has a interval which is exponent times bigger than the previous one

func (*DelayedConsumer) Start

func (dc *DelayedConsumer) Start() error

Start starts the subscription on the given connection

func (*DelayedConsumer) Stop

func (dc *DelayedConsumer) Stop() error

Stop stops the consumer

type Event

type Event struct {
	ConsentID            string  `json:"consentId"`
	TransactionID        string  `json:"transactionId"`
	InitiatorLegalEntity string  `gorm:"not null" json:"initiatorLegalEntity"`
	Error                *string `json:"error"`
	ExternalID           string  `gorm:"not null" json:"externalId"`
	Payload              string  `gorm:"not null" json:"payload"`
	RetryCount           int     `json:"retryCount"`
	Name                 string  `gorm:"not null" json:"name"`
	UUID                 string  `gorm:"PRIMARY_KEY" json:"uuid"`
}

Event is the type used for Gorm

func (Event) String

func (e Event) String() string

type EventHandlerCallback

type EventHandlerCallback func(event *Event)

EventHandlerCallback defines the signature of an event handler method.

type EventOctopus

type EventOctopus struct {
	Name   string
	Config EventOctopusConfig

	Db *gorm.DB
	// contains filtered or unexported fields
}

EventOctopus is the default implementation for EventOctopusInstance

func EventOctopusInstance

func EventOctopusInstance() *EventOctopus

EventOctopusInstance returns the EventOctopus singleton

func (*EventOctopus) Configure

func (octopus *EventOctopus) Configure() error

Configure initiates a STAN context

func (*EventOctopus) Diagnostics

func (octopus *EventOctopus) Diagnostics() []core.DiagnosticResult

Diagnostics returns diagnostic reports from the nats streaming service and the DB

func (*EventOctopus) EventPublisher

func (octopus *EventOctopus) EventPublisher(clientID string) (IEventPublisher, error)

EventPublisher gets a connection and creates a new EventPublisher

func (*EventOctopus) GetEvent

func (octopus *EventOctopus) GetEvent(uuid string) (*Event, error)

GetEvent returns single event or not based on given uuid

func (*EventOctopus) GetEventByExternalID

func (octopus *EventOctopus) GetEventByExternalID(externalID string) (*Event, error)

GetEventByExternalID returns single event or not based on given uuid

func (*EventOctopus) List

func (octopus *EventOctopus) List() (*[]Event, error)

List returns all current events from Db

func (*EventOctopus) RunMigrations

func (octopus *EventOctopus) RunMigrations(db *sql.DB) error

RunMigrations runs all new migrations in order

func (*EventOctopus) SaveOrUpdateEvent

func (octopus *EventOctopus) SaveOrUpdateEvent(event Event) error

SaveOrUpdateEvent saves or update the event in the store.

func (*EventOctopus) Shutdown

func (octopus *EventOctopus) Shutdown() error

Shutdown closes the connection to the DB and the natsServer server

func (*EventOctopus) Start

func (octopus *EventOctopus) Start() error

Start starts the receiver socket in a go routine

func (*EventOctopus) Subscribe

func (octopus *EventOctopus) Subscribe(service, subject string, handlers map[string]EventHandlerCallback) error

Subscribe lets you subscribe to events for a service and subject. For each Event.name you can provide a callback function

func (*EventOctopus) Unsubscribe

func (octopus *EventOctopus) Unsubscribe(service, subject string) error

Unsubscribe from a service and subject. If no subjects for a service are left, it closes the stanClient

type EventOctopusClient

type EventOctopusClient interface {
	EventPublisher(clientID string) (IEventPublisher, error)
	Subscribe(service, subject string, callbacks map[string]EventHandlerCallback) error
	Diagnostics() []core.DiagnosticResult
}

EventOctopusClient is the client interface for publishing events

type EventOctopusConfig

type EventOctopusConfig struct {
	RetryInterval      int
	NatsPort           int
	Connectionstring   string
	AutoRecover        bool
	PurgeCompleted     bool
	MaxRetryCount      int
	IncrementalBackoff int
}

EventOctopusConfig holds the config for the EventOctopusInstance

func (EventOctopusConfig) GetMode

func (c EventOctopusConfig) GetMode() string

GetMode derives the mode (from the global mode) the engine should run in

type EventPublisher

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

EventPublisher is a small wrapper around a natsClient so the user can pass an Event to Publish instead of a []byte

func (EventPublisher) Publish

func (p EventPublisher) Publish(subject string, event Event) error

Publish accepts an Event, than marshals and publishes it at the subject choice

type IEventPublisher

type IEventPublisher interface {
	Publish(subject string, event Event) error
}

IEventPublisher defines the Publish signature so it can be mocked or implemented for another tech