WIP Implement the metadata call to retrieve the stream lead (#37)

Add support for the enviroment

Author: Gsantomaggio

.github/workflows/test_and_publish_image.yml

@@ -10,7 +10,7 @@ jobs:
       rabbitmq-streaming:
         image: pivotalrabbitmq/rabbitmq-stream
         env:
-          RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS: "-rabbitmq_stream"
+          RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS: "-rabbitmq_stream  advertised_host localhost"
         ports:
           - 5551:5551
     steps:

Dockerfile

@@ -3,6 +3,7 @@ ENV GOPATH=/go GOOS=linux CGO_ENABLED=0
 WORKDIR /go/src/github.com/rabbitmq/rabbitmq-stream-go-client
 COPY go.mod go.sum VERSION ./
 COPY pkg pkg
+COPY Makefile Makefile
 COPY perfTest perfTest
 
 RUN mkdir /stream_perf_test

Makefile

@@ -10,8 +10,9 @@ LDFLAGS = "-X main.Version=$(VERSION)"
 
 all: test build
 
-vet:
-	go vet ./...
+
+vet: $(go_sources)
+	go vet ./pkg/stream
 
 fmt:
 	go fmt ./...
@@ -20,10 +21,13 @@ STATICCHECK ?= $(GOBIN)/staticcheck
 $(STATICCHECK):
 	go get honnef.co/go/tools/cmd/staticcheck
 check: $(STATICCHECK)
-	$(STATICCHECK) ./pkg/streaming
+	$(STATICCHECK) ./pkg/stream
 
 test: vet fmt check
-	go test -v  ./pkg/streaming -race -coverprofile=coverage.txt -covermode=atomic
+	go test -v  ./pkg/stream -race -coverprofile=coverage.txt -covermode=atomic
+
+integration-test: vet fmt check
+	go test -v  ./pkg/system_integration -race -coverprofile=coverage.txt -covermode=atomic -tags debug
 
 build: vet fmt check
 	go build -ldflags=$(LDFLAGS) -v ./...
@@ -39,4 +43,4 @@ perf-test-docker-build: perf-test-build
 	docker build -t pivotalrabbitmq/go-stream-perf-test:$(VERSION) .
 
 perf-test-docker-push: perf-test-docker-build
-	docker push pivotalrabbitmq/go-stream-perf-test:$(VERSION)
+	docker push pivotalrabbitmq/go-stream-perf-test:$(VERSION)

README.md

@@ -8,86 +8,80 @@ Experimental client for [RabbitMQ Stream Queues](https://github.com/rabbitmq/rab
 ### Download
 ---
 ```
-go get -u github.com/rabbitmq/rabbitmq-stream-go-client@v0.3-alpha
+go get -u github.com/rabbitmq/rabbitmq-stream-go-client@v0.4-alpha
 ```
 
-### How to test
+### Getting started
 ---
 - Run RabbitMQ docker image with streaming:
    ```
    docker run -it --rm --name rabbitmq -p 5551:5551 -p 5672:5672 -p 15672:15672 \
    -e RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="-rabbitmq_stream advertised_host localhost" \
    pivotalrabbitmq/rabbitmq-stream
- 
   ```
-- Run getting started example:
+- Run "getting started" example:
   ```
    go run examples/getting_started.go
   ```
-### Performance Test
----
-The performance tool is work in progress, you can use it with docker
-```
-docker run --network host -it pivotalrabbitmq/go-stream-perf-test silent 
-```
 
-or directly 
+### Performance test tool is an easy way to do some test:
 ```
-go run perfTest/perftest.go 
+go run perfTest/perftest.go silent
 ```
 
-
-
-
 ### API
 ---
 
+The API are generally composed by mandatory arguments and optional arguments
+the optional arguments can be set in the standard go way as:
 ```golang
-client, err := streaming.NewClientCreator().Uri(uris).Connect() // Create and Connect a client
+env, err := stream.NewEnvironment(
+            &stream.EnvironmentOptions{
+                    ConnectionParameters:  stream.Broker{
+                    Host:     "localhost",
+                    Port:     5551,
+                    User:     "guest",
+                    Password: "guest",
+                },
+                MaxProducersPerClient: 3,
+                MaxConsumersPerClient: 3,
+                },
+            )
 ```
-
-```golang
-err = client.StreamCreator().Stream(streamName).Create() // Create streaming queue without parameters
-err = client.StreamCreator().Stream(streamName).MaxAge(120 * time.Hour).Create() // Create streaming queue with Max Age
-err = client.StreamCreator().Stream(streamName).MaxLengthBytes(streaming.ByteCapacity{}.B(5)).Create() // Create streaming queue 5 GB max lenght
+or using Builders as:
 ```
-
-```golang
-/// Implement a consumer
-consumer, err := client.ConsumerCreator().
-		Stream(streamName).
-		Name("my_consumer").
-		MessagesHandler(func(context streaming.ConsumerContext, message *amqp.Message) {
-			fmt.Printf("received %d, message %s \n", context.Consumer.ID, message.Data)
-		}).Build()
+env, err := stream.NewEnvironment(
+		stream.NewEnvironmentOptions().
+			SetHost("localhost").
+			SetPort(5551).
+			SetUser("guest").
+			SetPassword("guest"))
 ```
 
+`nil` is also a valid value, default values will be provided:
 ```golang
-/// get a producer
-producer, err := client.ProducerCreator().Stream(streamName).Build()
+env, err := stream.NewEnvironment(nil) 
 ```
 
+The suggested way is to use builders.
+
+
 ### Build from source
 ---
 
 ```shell
 make build
 ```
 
+You need a docker image running to execute the tests in this way:
+```
+ docker run -it --rm --name rabbitmq -p 5551:5551 \
+   -e RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="-rabbitmq_stream advertised_host localhost" \
+   pivotalrabbitmq/rabbitmq-stream
+```
+
+
 
-### Methods Implemented:
----
- - Open(vhost)
- - CreateStream
- - DeleteStream
- - DeclarePublisher
- - Close Publisher
- - Publish
- - Subscribe 
- - Commit   
- - UnSubscribe
- - HeartBeat
- 
  ### Project status
  ---
  The client is a work in progress, the API(s) could change

VERSION

@@ -1 +1 @@
-0.4-alpha
+0.5-alpha

comopose/conf/rabbitmq.config

@@ -0,0 +1,4 @@
+[ { rabbit, [
+  {cluster_nodes, {['rabbit@rabbit_node_1', 'rabbit@rabbit_node_2'], disc}},
+  { loopback_users, [ ] } ] }
+].

comopose/docker-compose.yml

@@ -0,0 +1,42 @@
+version: "3"
+services:
+  rabbit_node_1:
+    environment:
+      - RABBITMQ_ERLANG_COOKIE='secret_cookie'
+    networks:
+      - back
+    hostname: rabbit_node_1
+    image: pivotalrabbitmq/rabbitmq-stream
+    ports:
+      - "15672:15672"
+      - "5551:5551"
+    tty: true
+    volumes:
+#      - rabbit1:/var/lib/rabbitmq
+      - ./conf/:/etc/rabbitmq/
+#    command:  bash -c "sleep 10; rabbitmq-server;"
+  rabbit_node_2:
+    environment:
+      - RABBITMQ_ERLANG_COOKIE='secret_cookie'
+    networks:
+      - back
+    hostname: rabbit_node_2
+    depends_on:
+      - rabbit_node_1
+    image: pivotalrabbitmq/rabbitmq-stream
+    ports:
+      - "15673:15672"
+      - "5552:5551"
+    tty: true
+    volumes:
+#      - rabbit2:/var/lib/rabbitmq
+      - ./conf/:/etc/rabbitmq/
+#    command:  bash -c "sleep 10; rabbitmq-server;"
+#volumes:
+#  rabbit1:
+#    driver: local
+#  rabbit2:
+#    driver: local
+
+networks:
+  back:

examples/getting_started.go

@@ -6,96 +6,90 @@ import (
 	"fmt"
 	"github.com/google/uuid"
 	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/amqp"
-	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/streaming"
+	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
 	"os"
-	"sync/atomic"
+	"strconv"
 	"time"
 )
 
 func CheckErr(err error) {
 	if err != nil {
-		streaming.ERROR("%s ", err)
+		fmt.Printf("%s ", err)
+		os.Exit(1)
 	}
 }
+
+func CreateArrayMessagesForTesting(bacthMessages int) []*amqp.Message {
+	var arr []*amqp.Message
+	for z := 0; z < bacthMessages; z++ {
+		arr = append(arr, amqp.NewMessage([]byte("hello_world_"+strconv.Itoa(z))))
+	}
+	return arr
+}
+
 func main() {
 	reader := bufio.NewReader(os.Stdin)
-	streaming.INFO("Getting started with Streaming client for RabbitMQ")
-	streaming.INFO("Connecting to RabbitMQ streaming ...")
-	uris := "rabbitmq-streaming://guest:guest@localhost:5551/%2f"
-	client, err := streaming.NewClientCreator().
-		Uri(uris).
-		Connect() // Create Client
-	CheckErr(err)
-	if err != nil {
-		return
-	}
+	// Set log level, not mandatory by default is INFO
+	stream.SetLevelInfo(stream.DEBUG)
 
-	streaming.INFO("Connected to: %s", uris)
-	streamName := uuid.New().String()
-	err = client.StreamCreator().Stream(streamName).
-		Create() // Create the streaming queue
-	CheckErr(err)
+	fmt.Println("Getting started with Streaming client for RabbitMQ")
+	fmt.Println("Connecting to RabbitMQ streaming ...")
+	//uri := "rabbitmq-streaming://guest:guest@localhost:5551/%2f"
+	// The environment is a wrapper around the TCP client connections
 
-	err = client.StreamCreator().Stream(streamName).
-		MaxLengthBytes(streaming.ByteCapacity{}.MB(5)).
-		Create() // Create the streaming queue
+	env, err := stream.NewEnvironment(
+		stream.NewEnvironmentOptions().
+			SetHost("localhost").
+			SetPort(5551).
+			SetUser("guest").
+			SetPassword("guest"))
 	CheckErr(err)
+	// Create a stream, you can create streams without any option like:
+	// err = env.DeclareStream(streamName, nil)
+	// it is a best practise to define a size,  1GB for example:
+	streamName := uuid.New().String()
+	err = env.DeclareStream(streamName,
+		&stream.StreamOptions{
+			MaxLengthBytes: stream.ByteCapacity{}.GB(2),
+		},
+	)
 
-	var count int32
-	consumer, err := client.ConsumerCreator().
-		Stream(streamName).
-		Name(uuid.NewString()).
-		MessagesHandler(func(context streaming.ConsumerContext, message *amqp.Message) {
-			streaming.INFO("Message number:%d consumer id:%d data:%s \n",
-				atomic.AddInt32(&count, 1), context.Consumer.ID,
-				message.Data)
-			err := context.Consumer.Commit()
-			CheckErr(err)
-		}).Build()
 	CheckErr(err)
 
-	// Get a new producer to publish the messages
-	clientProducer, err := streaming.NewClientCreator().Uri(uris).
-		PublishErrorHandler(func(publisherId uint8, publishingId int64, code uint16) {
-			streaming.ERROR("Publish Error, publisherId %d, code: %s", publisherId, streaming.LookErrorCode(code))
-		}).
-		Connect()
-	CheckErr(err)
-	producer, err := clientProducer.ProducerCreator().Stream(streamName).Build()
-	CheckErr(err)
+	//Define a producer to a stream, optional publish confirmation
+	producer, err := env.NewProducer(streamName,
+		stream.NewProducerOptions().SetPublishConfirmHandler(func(ch <-chan []int64) {
+			messagesIds := <-ch
+			fmt.Printf("Confirmed %d messages \n \n ", len(messagesIds))
 
-	//
-	numberOfSend := 10
-	batchSize := 10
+		}))
+	CheckErr(err)
 
-	// Create AMQP 1.0 messages, see:https://github.com/Azure/go-amqp
-	// message aggregation
-	countM := 0
-	start := time.Now()
-	for z := 0; z < numberOfSend; z++ {
-		var arr []*amqp.Message
-		for f := 0; f < batchSize; f++ {
-			countM++
-			arr = append(arr, amqp.NewMessage([]byte(fmt.Sprintf("test_%d", countM))))
-		}
-		_, err = producer.BatchPublish(context.Background(), arr) // batch send
-		if err != nil {
-			streaming.ERROR("%s", err)
-		}
+	// each publish sends a number of messages, the batchMessages should be around 100 messages for send
+	for i := 0; i < 2; i++ {
+		_, err := producer.BatchPublish(context.Background(), CreateArrayMessagesForTesting(10))
+		CheckErr(err)
 	}
 
-	elapsed := time.Since(start)
-	streaming.INFO("%d messages, published in: %s\n", numberOfSend*batchSize, elapsed)
-
-	fmt.Println("Press any key to stop ")
-	_, _ = reader.ReadString('\n')
+	// this sleep is not mandatory, just to show the confirmed messages
+	time.Sleep(1 * time.Second)
 	err = producer.Close()
 	CheckErr(err)
-	err = consumer.UnSubscribe()
-	CheckErr(err)
-	err = client.DeleteStream(streamName) // Remove the streaming queue and the data
+
+	// Define a consumer per stream, there are different offset options to define a consumer, default is
+	//env.NewConsumer(streamName, func(Context streaming.ConsumerContext, message *amqp.Message) {
+	//
+	//}, nil)
+	// if you need to track the offset you need a consumer name like:
+	consumer, err := env.NewConsumer(streamName, func(Context stream.ConsumerContext, message *amqp.Message) {
+		fmt.Printf("consumer id: %d, text: %s \n ", Context.Consumer.ID, message.Data)
+	}, stream.NewConsumerOptions().
+		SetConsumerName("my_consumer").                  // gives a name
+		SetOffset(stream.OffsetSpecification{}.First())) // start consuming from the beginning
 	CheckErr(err)
-	err = client.Close()
+
+	fmt.Println("Press any key to stop ")
+	err = consumer.UnSubscribe()
 	CheckErr(err)
-	fmt.Println("Bye bye")
+	_, _ = reader.ReadString('\n')
 }