Initial implementation of distributed topic

dtopic-client.go

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/buraksezer/olric"
	"github.com/buraksezer/olric/client"
	"github.com/buraksezer/olric/serializer"
)

func main() {
	cc := &client.Config{
		Addrs:      []string{"192.168.1.4:3320"},
		MaxConn:    10,
		Serializer: serializer.NewMsgpackSerializer(),
	}

	// Create a new client instance
	c, err := client.New(cc)
	if err != nil {
		log.Fatalf("Olric client returned error: %v", err)
	}
	defer c.Close()

	// Create a DTopic instance
	dt, err := c.NewDTopic("my-dtopic", 0, olric.UnorderedDelivery)
	if err != nil {
		log.Fatalf("Olric client returned error: %v", err)
	}

	ctx, cancel := context.WithCancel(context.Background())
	onMessage := func(msg olric.DTopicMessage) {
		defer cancel()
		fmt.Println("Message: ", msg.Message)
		fmt.Println("PublisherAddr:", msg.PublisherAddr)
		fmt.Println("PublishedAt:", msg.PublishedAt)
	}

	listenerID, err := dt.AddListener(onMessage)
	if err != nil {
		log.Fatalf("DTopic.AddListener returned an error: %v", err)
	}

	defer func() {
		err = dt.RemoveListener(listenerID)
		if err != nil {
			log.Fatalf("DTopic.RemoveListener returned an error: %v", err)
		}
	}()

	err = dt.Publish("Hello World!")
	if err != nil {
		log.Fatalf("DTopic.Publish returned an error: %v", err)
	}

	select {
	case <-ctx.Done():
		log.Println("Good bye!")
	}
}