eventor

README

eventor

🔮 A minimalistic library for abstracting pub/sub operations

→ eventor is clerk for pub/sub 😉

Installation

go get github.com/Becklyn/eventor

Supported brokers

eventor has builtin support for the following brokers:

• Dapr Pub/sub API - APIs for building portable and reliable microservices

Usage

Being a minimalistic libary, eventor only provides you with the basiscs. The rest is up to your specific need.

Creating a publisher

publisher := dapr.NewDaprPublisher(dapr.DaprPublisherConfig{
    Hostname: "localhost",
    Port:     "3500",
    Pubsub:   "redis-pubsub",
})

Creating a subscriber registry

subscriber := dapr.NewDaprSubscriberRegistry(dapr.DaprSubscriberRegistryConfig{
    App:    app,
    Pubsub: "redis-pubsub",
})

Publish

type Message struct {
    Id string `json:"id"`
    Body string `json:"body"`
}

if err := publisher.Publish("topic", Message{
    Id:   "0",
    Body: "Hello World",
}); err != nil {
    panic(err)
}

Subscribe with handler function

type Message struct {
    Id string `json:"id"`
    Body string `json:"body"`
}

handlerFn := func(msg Message) error {
    fmt.Println(msg)
}

usubscribe, err := eventor.On(handlerFn, subscriber, "topic")
if err != nil {
    panic(err)
}
defer unsubscribe()

Subscribe with channel

type Message struct {
    Id string `json:"id"`
    Body string `json:"body"`
}

messageChan, unsubscribe, err := eventor.Chan[Message](subscriber, "topic")
if err != nil {
    panic(err)
}
defer unsubscribe()

message := <-messageChan
fmt.Println(message)

Documentation

Index

• type HandlerFn
• type Publisher
• type Subscriber
• type UnsubscribeFn
  • func Chan[T any](subscriber Subscriber, topic string, match ...func(event T) bool) (<-chan T, UnsubscribeFn, error)
  • func On[T any](handlerFn func(event T, spanContext context.Context) error, ...) (UnsubscribeFn, error)

Types

type HandlerFn

type HandlerFn func(event cloudevents.Event, parentSpanCtx context.Context) error

HandlerFn is a function that is called when a cloudevent is received at a subscriber.

type Publisher

type Publisher interface {
	// Publish publishes any data to the given topic.
	// A topic needs to be passed to which the publisher publisheds the data.
	// The data can be of any arbitrary type.
	Publish(topic string, data any, spanCtx ...trace.SpanContext) error
}

type Subscriber

type Subscriber interface {
	// Subscribe registers a subscriber at the registry.
	// A topic needs to be passed from which the subscriber will receive events.
	// The handler function is called when an event is received.
	Subscribe(topic string, handler HandlerFn) error

	// Unsubscribe removes a subscriber from the registry.
	// The handler function of the handler to remove has to be passed.
	Unsubscribe(handler HandlerFn) error
}

type UnsubscribeFn

type UnsubscribeFn func()

UnsubscribeFn is a function to unsubscribe a handler.

func Chan

func Chan[T any](subscriber Subscriber, topic string, match ...func(event T) bool) (<-chan T, UnsubscribeFn, error)

Chan is a convenience function for creating a channel to an event handler. A subscriber has to be passed on which the internal handler gets registered. A topic needs to be passed from which the subscriber will receive events. A match function can be passed to filter events, reaching the channel. The method returns a channel and a function to unsubscribe the handler.

func On

func On[T any](handlerFn func(event T, spanContext context.Context) error, subscriber Subscriber, topic string) (UnsubscribeFn, error)

On is a convenience function for creating a handler that executes only if the type can be parsed. A subscriber has to be passed on which the internal handler gets registered. A topic needs to be passed from which the subscriber will receive events. The method returns a function to unsubscribe the handler.