messaging

type EventObservable interface {
	AddListener(common.EventType, common.TransactionID, Listener) (ListenerID, error)
	// AddListenerMultiEventType register a listener to a set of common.EventType
	AddListenerMultiEventType([]common.EventType, common.TransactionID, Listener) (ListenerID, error)
	RemoveListenerByID(id ListenerID)
}

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

type Listener struct {
	Callback ListenerCallback
	// if enabled, then listener will be removed after been called once.
	// default to false.
	ListenOnce bool
}

type ListenerCallback func(ev common.EventType, ce cloudevents.Event)

type ListenerID string

type MockNatsConnection struct {
	Config                   *NatsConfig
	CommonSubjectRegex       *regexp.Regexp
	HandlerLock              sync.Mutex
	EventHandlers            map[common.QueryOp]QueryEventHandler
	DefaultEventHandler      QueryEventHandler
	CloudEventHandlers       map[common.QueryOp]QueryCloudEventHandler
	DefaultCloudEventHandler QueryCloudEventHandler
}

type MockNatsMessage struct {
	Data         []byte
	ResponseData []byte
}

type MockStanConnection struct {
	NatsConfig               *NatsConfig
	StanConfig               *StanConfig
	HandlerLock              sync.Mutex
	EventHandlers            map[common.EventType]StreamingEventHandler
	DefaultEventHandler      StreamingEventHandler
	CloudEventHandlers       map[common.EventType]StreamingCloudEventHandler
	DefaultCloudEventHandler StreamingCloudEventHandler
}

type NatsConfig struct {
	URL             string `envconfig:"NATS_URL" default:"nats://nats:4222"`
	QueueGroup      string `envconfig:"NATS_QUEUE_GROUP"`
	WildcardSubject string `envconfig:"NATS_WILDCARD_SUBJECT" default:"cyverse.>"` // WildcardSubject field is optional, only used for NATS Query
	ClientID        string `envconfig:"NATS_CLIENT_ID"`                            // While not strictly used by Nats, this is propagated into the cloudevent
	MaxReconnects   int    `envconfig:"NATS_MAX_RECONNECTS" default:"-1"`          // implementation should default to DefaultNatsMaxReconnect
	ReconnectWait   int    `envconfig:"NATS_RECONNECT_WAIT" default:"-1"`          // in seconds, implementation should default to
	RequestTimeout  int    `envconfig:"NATS_REQUEST_TIMEOUT" default:"-1"`
}

type NatsConnection struct {
	Config                   *NatsConfig
	Connection               *nats.Conn
	Subscription             *nats.Subscription
	HandlerLock              sync.Mutex
	EventHandlers            map[common.QueryOp]QueryEventHandler
	DefaultEventHandler      QueryEventHandler
	CloudEventHandlers       map[common.QueryOp]QueryCloudEventHandler
	DefaultCloudEventHandler QueryCloudEventHandler
}

type QueryCloudEventHandler func(event *cloudevents.Event) ([]byte, error)

type QueryEventHandler func(subject common.QueryOp, transactionID common.TransactionID, jsonData []byte) ([]byte, error)

type QueryEventHandlerMapping struct {
	Subject common.QueryOp // required, empty string for default handler

	// optional, set CloudEventHandler or EventHandler
	CloudEventHandler QueryCloudEventHandler
	EventHandler      QueryEventHandler
}

type QueryEventService interface {
	Disconnect() error

	AddEventHandler(subject common.QueryOp, eventHandler QueryEventHandler) error
	AddDefaultEventHandler(eventHandler QueryEventHandler) error
	AddCloudEventHandler(subject common.QueryOp, eventHandler QueryCloudEventHandler) error
	AddDefaultCloudEventHandler(eventHandler QueryCloudEventHandler) error
	Request(ctx context.Context, subject common.QueryOp, data interface{}) ([]byte, error)
	RequestWithTransactionID(ctx context.Context, subject common.QueryOp, data interface{}, transactionID common.TransactionID) ([]byte, error)
	RequestCloudEvent(ctx context.Context, ce *cloudevents.Event) ([]byte, error)
}

type StanConfig struct {
	ClusterID     string `envconfig:"NATS_CLUSTER_ID" default:"cacao-cluster"`
	DurableName   string `envconfig:"NATS_DURABLE_NAME"`
	EventsTimeout int    `envconfig:"NATS_EVENTS_TIMEOUT" default:"-1"`
}

type StanConnection struct {
	NatsConfig               *NatsConfig
	StanConfig               *StanConfig
	Connection               stan.Conn
	Subscription             stan.Subscription
	HandlerLock              sync.Mutex
	EventHandlers            map[common.EventType]StreamingEventHandler
	DefaultEventHandler      StreamingEventHandler
	CloudEventHandlers       map[common.EventType]StreamingCloudEventHandler
	DefaultCloudEventHandler StreamingCloudEventHandler
}

type StreamingCloudEventHandler func(event *cloudevents.Event) error

type StreamingEventHandler func(subject common.EventType, transactionID common.TransactionID, jsonData []byte) error

type StreamingEventHandlerMapping struct {
	Subject common.EventType // required, empty string for default handler

	// optional, set CloudEventHandler or EventHandler
	CloudEventHandler StreamingCloudEventHandler
	EventHandler      StreamingEventHandler
}

type StreamingEventService interface {
	Disconnect() error

	AddEventHandler(subject common.EventType, eventHandler StreamingEventHandler) error
	AddDefaultEventHandler(eventHandler StreamingEventHandler) error
	AddCloudEventHandler(subject common.EventType, eventHandler StreamingCloudEventHandler) error
	AddDefaultCloudEventHandler(eventHandler StreamingCloudEventHandler) error
	Publish(subject common.EventType, data interface{}) error
	PublishWithTransactionID(subject common.EventType, data interface{}, transactionID common.TransactionID) error
	PublishCloudEvent(ce *cloudevents.Event) error
}