messaging2

type EventConnection interface {
	// Listen listens for some events and call the corresponding handler when
	// receiving those events. This should only be called once on the
	// EventConnection.
	Listen(context.Context, map[common.EventType]EventHandlerFunc, *sync.WaitGroup) error

	// ListenWithConcurrentWorkers will start listening for events, just like
	// Listen(), but this will spawn worker go-routines that consume from an internal
	// channel. This allows for bounded concurrency. Channel buffer and worker count
	// need to be passed when calling this function.
	ListenWithConcurrentWorkers(ctx context.Context, m map[common.EventType]EventHandlerFunc, wg *sync.WaitGroup, channelBufferLen uint, workerCount uint) error

	// Publish will publish a cloudevent, the cloudevent source will be automatically populated if it is empty or AutoPopulateCloudEventSource.
	Publish(ce cloudevents.Event) error
	Request(requestCe cloudevents.Event, responseEventTypes []common.EventType) (EventResponsePromise, error)
	WaitForResponse(transactionID common.TransactionID, responseEventTypes []common.EventType) (EventResponsePromise, error)
}

type EventHandlerFunc func(ctx context.Context, event cloudevents.Event, writer EventResponseWriter) error

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 EventResponsePromise interface {
	// Response wait for response
	Response(ctx context.Context) (cloudevents.Event, error)
	// ResponseChan returns a channel that will be receiving the responses
	ResponseChan(ctx context.Context) <-chan cloudevents.Event
	// Close cleans up any resources used by the promise, this is useful when you do
	// not want to receive the response, or when response never arrives.
	//
	// If response is successfully received, then there should be NO need to call Close()
	Close() error
}

type EventResponseWriter interface {
	// Write will publish the cloudevent, if the transaction ID is empty it will be
	// overridden with the transaction ID associated with the Incoming event that
	// triggered EventHandlerFunc. Write can be called multiple times on the same EventResponseWriter.
	Write(ce cloudevents.Event) error
	// NoResponse explicitly indicate that there is no response
	NoResponse()
}

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 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 is used for Nats Client Name, STAN Client ID, cloudevent source.
	ClientID       string `envconfig:"NATS_CLIENT_ID"`
	MaxReconnects  int    `envconfig:"NATS_MAX_RECONNECTS" default:"-1"`  // in seconds, implementation should default to DefaultNatsMaxReconnect
	ReconnectWait  int    `envconfig:"NATS_RECONNECT_WAIT" default:"-1"`  // in seconds, implementation should default to DefaultNatsReconnectWait
	RequestTimeout int    `envconfig:"NATS_REQUEST_TIMEOUT" default:"-1"` // in seconds, implementation should default to DefaultNatsRequestTimeout

	// EventWildcardSubject field is optional, only used for NATS JetStream Events
	EventWildcardSubject string `envconfig:"NATS_JS_WILDCARD_SUBJECT"`
	// AckWaitSec is the seconds to wait for acknowledgement or time the messaging
	// bus wait before attempting redelivery, this should be set to the maximum
	// seconds it takes for service to process an operation.
	AckWaitSec int `envconfig:"NATS_JS_ACK_WAIT" default:"-1"` // in seconds, implementation should default to DefaultAckWait
}

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

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

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

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

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

type NatsStanMsgConfig struct {
	NatsConfig
	StanConfig
}

type QueryConnection interface {
	Request(ctx context.Context, requestCe cloudevents.Event) (reply cloudevents.Event, err error)
	// RequestStream request for multiple replies.
	// An example of the implementation is Scatter-Gather pattern.
	// https://docs.nats.io/using-nats/developer/sending/request_reply#scatter-gather
	RequestStream(ctx context.Context, requestCe cloudevents.Event) (ReplyStream, error)
	// Listen will start listening for queries, and call corresponding handlers if
	// query type matches what is provided in the map.
	//
	// Caller can stop listening by cancel the context passed to Listen(). WaitGroup
	// is for waiting for the Listen to stop, since stop listening is not immediate.
	//
	// Note that this is a non-blocking call, as in it does not block until the
	// listen stopped, use the WaitGroup to wait for Listen to be stopped.
	Listen(context.Context, map[common.QueryOp]QueryHandlerFunc, *sync.WaitGroup) error
	// ListenWithConcurrentWorkers will start listening for queries, just like
	// Listen(), but this will spawn worker go-routines that consume from an internal
	// channel. This allows for bounded concurrency. Channel buffer and worker count
	// need to be passed when calling this function.
	ListenWithConcurrentWorkers(ctx context.Context, m map[common.QueryOp]QueryHandlerFunc, wg *sync.WaitGroup, channelBufferLen uint, workerCount uint) error
}

type QueryHandlerFunc func(ctx context.Context, request cloudevents.Event, writer ReplyWriter)

type ReplyStream interface {
	ExpectNext() bool
	Next() (cloudevents.Event, error)
}

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

type ReplyWriter interface {
	Write(cloudevents.Event) error
}

type StanConfig struct {
	// ClusterID is STAN specific
	ClusterID string `envconfig:"NATS_CLUSTER_ID" default:"cacao-cluster"`
	// DurableName is STAN/JetStream specific
	DurableName   string `envconfig:"NATS_DURABLE_NAME"`
	EventsTimeout int    `envconfig:"NATS_EVENTS_TIMEOUT" default:"-1"` // in seconds, implementation should default to DefaultStanEventsTimeout
}

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

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

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