Add stream tutorial 2 for Go files

go-stream/offset_tracking_receive.go

@@ -0,0 +1,66 @@
+package main
+
+import (
+	"fmt"
+	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/amqp"
+	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
+	"os"
+	"sync/atomic"
+)
+
+func main() {
+	env, err := stream.NewEnvironment(
+		stream.NewEnvironmentOptions().
+			SetHost("localhost").
+			SetPort(5552).
+			SetUser("guest").
+			SetPassword("guest"))
+	CheckErrReceive(err)
+
+	streamName := "stream-offset-tracking-go"
+	err = env.DeclareStream(streamName,
+		&stream.StreamOptions{
+			MaxLengthBytes: stream.ByteCapacity{}.GB(2),
+		},
+	)
+	CheckErrReceive(err)
+
+	var firstOffset, messageCount int64 = -1, 0
+	var lastOffset atomic.Int64
+	ch := make(chan bool)
+	messagesHandler := func(consumerContext stream.ConsumerContext, message *amqp.Message) {
+		if atomic.CompareAndSwapInt64(&firstOffset, -1, consumerContext.Consumer.GetOffset()) {
+			fmt.Println("First message received.")
+		}
+		if atomic.AddInt64(&messageCount, 1)%10 == 0 {
+			consumerContext.Consumer.StoreOffset()
+			fmt.Println("storing offset")
+		}
+		if string(message.GetData()) == "marker" {
+			lastOffset.Store(consumerContext.Consumer.GetOffset())
+			consumerContext.Consumer.StoreOffset()
+			consumerContext.Consumer.Close()
+			ch <- true
+		}
+	}
+
+	_, err = env.NewConsumer(streamName, messagesHandler,
+		stream.NewConsumerOptions().
+			SetManualCommit().
+			SetConsumerName("offset-tracking-tutorial").
+			SetOffset(stream.OffsetSpecification{}.LastConsumed()))
+	CheckErrReceive(err)
+	fmt.Println("Started consuming...")
+	_ = <-ch
+
+	fmt.Printf("Done consuming, first offset %d, last offset %d.\n", firstOffset, lastOffset.Load())
+	CheckErrReceive(err)
+
+}
+
+func CheckErrReceive(err error) {
+	if err != nil {
+		fmt.Printf("%s ", err)
+		os.Exit(1)
+	}
+}

go-stream/offset_tracking_send.go

@@ -0,0 +1,72 @@
+package main
+
+import (
+	"fmt"
+	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/amqp"
+	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
+	"os"
+)
+
+func main() {
+	env, err := stream.NewEnvironment(
+		stream.NewEnvironmentOptions().
+			SetHost("localhost").
+			SetPort(5552).
+			SetUser("guest").
+			SetPassword("guest"))
+	CheckErrSend(err)
+
+	streamName := "stream-offset-tracking-go"
+	err = env.DeclareStream(streamName,
+		&stream.StreamOptions{
+			MaxLengthBytes: stream.ByteCapacity{}.GB(2),
+		},
+	)
+	CheckErrSend(err)
+
+	producer, err := env.NewProducer(streamName, stream.NewProducerOptions())
+	CheckErrSend(err)
+
+	messageCount := 100
+	ch := make(chan bool)
+	chPublishConfirm := producer.NotifyPublishConfirmation()
+	handlePublishConfirm(chPublishConfirm, messageCount, ch)
+
+	fmt.Printf("Publishing %d messages\n", messageCount)
+	for i := 0; i < messageCount; i++ {
+		var body string
+		if i == messageCount-1 {
+			body = "marker"
+		} else {
+			body = "hello"
+		}
+		err = producer.Send(amqp.NewMessage([]byte(body)))
+		CheckErrSend(err)
+	}
+	_ = <-ch
+	fmt.Println("Messages confirmed")
+
+	err = producer.Close()
+	CheckErrSend(err)
+}
+
+func handlePublishConfirm(confirms stream.ChannelPublishConfirm, messageCount int, ch chan bool) {
+	go func() {
+		confirmedCount := 0
+		for confirmed := range confirms {
+			for _, _ = range confirmed {
+				confirmedCount++
+				if confirmedCount == messageCount {
+					ch <- true
+				}
+			}
+		}
+	}()
+}
+
+func CheckErrSend(err error) {
+	if err != nil {
+		fmt.Printf("%s ", err)
+		os.Exit(1)
+	}
+}