pubsub

README

PubSub

Installation

go get bitbucket.org/amotus/pubsub

Usage

package main

import "bitbucket.org/amotus/pubsub"

func main() {
	kafka = kafka.NewPubSub("redpanda:9092", "kafka-consumer-name-example", "topic")
	
	go kafka.Listen()
	
	topicChanOut := kafka.Subscribe("topic")

	go func() {
		for msg := range topicChanOut {
			onKafkaMessage(msg)
		}
	}()
}


func onKafkaMessage(msg kafka.KafkaMessage) {
	log.Println("received message", msg.Topic, msg.Payload)
}

TODO

• Update readme to include Blocking-pubsub example

License

Licensed under Apache License, Version 2.0 LICENSE.

Documentation

Overview

Package pubsub provide the implementation of a PubSub pattern with channels

Index

• func Consume[Input, Output any](ctx context.Context, input <-chan Input, output chan<- Output, ...)
• func Forward[Message1, Message2 any](subscriber <-chan Message1, broadcaster PubSub[Message2], ...)
• type MessageTransformer
• type PubSub
  • func NewBlockingPubSub[Message any]() PubSub[Message]
  • func NewPubSub[Message any]() PubSub[Message]

Functions

func Consume

func Consume[Input, Output any](ctx context.Context,
	input <-chan Input,
	output chan<- Output,
	transformer MessageTransformer[Input, Output],
	gate channel.MessageFilter[Output])

func Forward

func Forward[Message1, Message2 any](subscriber <-chan Message1, broadcaster PubSub[Message2], transform func(message Message1) Message2)

Types

type MessageTransformer

type MessageTransformer[IN any, OUT any] interface {
	Transform(message IN) OUT
}

type PubSub

type PubSub[Message any] interface {
	// Publish a message to a topic
	Publish(message Message) error

	// Subscribe to a topic
	Subscribe(filter channel.MessageFilter[Message]) (<-chan Message, error)

	// Unsubscribe from a topic
	Unsubscribe(<-chan Message) error
}

func NewBlockingPubSub

func NewBlockingPubSub[Message any]() PubSub[Message]

func NewPubSub

func NewPubSub[Message any]() PubSub[Message]