Consumer replica conn (#42)

* Add consumer broker replica
* add URIs configuration
* ha producer - first implementatio

Author: Gsantomaggio

.gitignore

@@ -20,3 +20,4 @@ examples/examples
 vet
 .DS_Store
 perfTest/
+.vagrant/

Makefile

@@ -26,7 +26,7 @@ test: vet fmt check
 	go test -v  ./pkg/stream -race -coverprofile=coverage.txt -covermode=atomic -tags debug
 
 integration-test: vet fmt check
-	go test -v  ./pkg/system_integration -race -coverprofile=coverage.txt -covermode=atomic -tags debug
+	cd ./pkg/system_integration && go test -v  . -race -coverprofile=coverage.txt -covermode=atomic -tags debug -timeout 99999s
 
 build: vet fmt check
 	go build -ldflags=$(LDFLAGS) -v ./...

README.md

@@ -3,16 +3,19 @@
 ![Build](https://github.com/rabbitmq/rabbitmq-stream-go-client/workflows/Build/badge.svg)
 [![codecov](https://codecov.io/gh/Gsantomaggio/go-stream-client/branch/main/graph/badge.svg?token=HZD4S71QIM)](https://codecov.io/gh/Gsantomaggio/go-stream-client)
 
-Experimental client for [RabbitMQ Stream Queues](https://github.com/rabbitmq/rabbitmq-server/tree/master/deps/rabbitmq_stream)
+Experimental client
+for [RabbitMQ Stream Queues](https://github.com/rabbitmq/rabbitmq-server/tree/master/deps/rabbitmq_stream)
 
 ### Download
 ---
+
 ```
 go get -u github.com/rabbitmq/rabbitmq-stream-go-client@v0.5-alpha
 ```
 
 ### Getting started
 ---
+
 - Run RabbitMQ docker image with streaming:
    ```
    docker run -it --rm --name rabbitmq -p 5552:5552 -p 5672:5672 -p 15672:15672 \
@@ -25,15 +28,17 @@ go get -u github.com/rabbitmq/rabbitmq-stream-go-client@v0.5-alpha
   ```
 
 ### Performance test tool is an easy way to do some test:
+
 ```
 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:
+The API are generally composed by mandatory arguments and optional arguments the optional arguments can be set in the
+standard go way as:
+
 ```golang
 env, err := stream.NewEnvironment(
             &stream.EnvironmentOptions{
@@ -48,7 +53,9 @@ env, err := stream.NewEnvironment(
                 },
             )
 ```
+
 or using Builders as:
+
 ```golang
 env, err := stream.NewEnvironment(
 		stream.NewEnvironmentOptions().
@@ -59,6 +66,7 @@ env, err := stream.NewEnvironment(
 ```
 
 `nil` is also a valid value, default values will be provided:
+
 ```golang
 env, err := stream.NewEnvironment(nil) 
 ```
@@ -74,14 +82,13 @@ make build
 ```
 
 You need a docker image running to execute the tests in this way:
+
 ```
  docker run -it --rm --name rabbitmq -p 5552:5552 \
    -e RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="-rabbitmq_stream advertised_host localhost" \
    pivotalrabbitmq/rabbitmq-stream
 ```
 
-
-
- ### Project status
- ---
- The client is a work in progress, the API(s) could change
+### Project status
+---
+The client is a work in progress, the API(s) could change

VERSION

@@ -1 +1 @@
-0.5-alpha
+0.6-alpha

comopose/conf/enabled_plugins

@@ -0,0 +1 @@
+[rabbitmq_management, rabbitmq_stream, rabbitmq_stream_management].

comopose/conf/rabbitmq.conf

@@ -0,0 +1,6 @@
+cluster_formation.peer_discovery_backend  = rabbit_peer_discovery_classic_config
+
+cluster_formation.classic_config.nodes.1 = rabbit@node0
+cluster_formation.classic_config.nodes.2 = rabbit@node1
+cluster_formation.classic_config.nodes.3 = rabbit@node2
+loopback_users.guest = false

comopose/conf/rabbitmq.config

@@ -1,42 +1,51 @@
 version: "3"
 services:
-  rabbit_node_1:
+  rabbit_node0:
     environment:
       - RABBITMQ_ERLANG_COOKIE='secret_cookie'
+      - RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS=-rabbitmq_stream advertised_host localhost advertised_port 5552
     networks:
       - back
-    hostname: rabbit_node_1
+    hostname: node0
     image: pivotalrabbitmq/rabbitmq-stream
     ports:
       - "15672:15672"
       - "5552:5552"
     tty: true
     volumes:
-#      - rabbit1:/var/lib/rabbitmq
+     - ./conf/:/etc/rabbitmq/
+  rabbit_node1:
+    environment:
+      - RABBITMQ_ERLANG_COOKIE='secret_cookie'
+      - RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS=-rabbitmq_stream advertised_host localhost advertised_port 5553
+    networks:
+      - back
+    hostname: node1
+    depends_on:
+      - rabbit_node_0
+    image: pivotalrabbitmq/rabbitmq-stream
+    ports:
+      - "15673:15672"
+      - "5553:5552"
+    tty: true
+    volumes:
       - ./conf/:/etc/rabbitmq/
-#    command:  bash -c "sleep 10; rabbitmq-server;"
-  rabbit_node_2:
+  rabbit_node2:
     environment:
       - RABBITMQ_ERLANG_COOKIE='secret_cookie'
+      - RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS=-rabbitmq_stream advertised_host localhost advertised_port 5554
     networks:
       - back
-    hostname: rabbit_node_2
+    hostname: node2
     depends_on:
       - rabbit_node_1
     image: pivotalrabbitmq/rabbitmq-stream
     ports:
-      - "15673:15672"
+      - "15674:15672"
       - "5554:5552"
     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:

comopose/docker-compose.yml

@@ -29,9 +29,10 @@ func CreateArrayMessagesForTesting(bacthMessages int) []*amqp.Message {
 
 func handlePublishConfirm(confirms stream.ChannelPublishConfirm) {
 	go func() {
-		messagesIds := <-confirms
-		for _, m := range messagesIds {
-			fmt.Printf("Confirmed %s message \n  ", m.Message.Data)
+		for messagesIds := range confirms {
+			for _, m := range messagesIds {
+				fmt.Printf("Confirmed %s message \n  ", m.Message.Data)
+			}
 		}
 	}()
 }
@@ -59,6 +60,7 @@ func main() {
 	// 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{

examples/getting_started.go

@@ -0,0 +1,114 @@
+package main
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/amqp"
+	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/ha"
+	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
+	"os"
+	"strconv"
+	"sync/atomic"
+	"time"
+)
+
+func CheckErr(err error) {
+	if err != nil {
+		fmt.Printf("%s ", err)
+		os.Exit(1)
+	}
+}
+
+var idx = 0
+
+func CreateArrayMessagesForTesting(bacthMessages int) []*amqp.Message {
+	var arr []*amqp.Message
+	for z := 0; z < bacthMessages; z++ {
+		idx++
+		arr = append(arr, amqp.NewMessage([]byte(strconv.Itoa(idx))))
+	}
+	return arr
+}
+
+func handlePublishConfirm(confirms stream.ChannelPublishConfirm) {
+	var counter int32 = 0
+	go func() {
+		for messagesIds := range confirms {
+			for _, m := range messagesIds {
+				if !m.Confirmed {
+					if atomic.AddInt32(&counter, 1)%10 == 0 {
+						fmt.Printf("Confirmed %s message - status %t - %d \n  ", m.Message.Data, m.Confirmed, atomic.LoadInt32(&counter))
+					}
+				}
+			}
+		}
+	}()
+}
+
+func main() {
+	reader := bufio.NewReader(os.Stdin)
+	// Set log level, not mandatory by default is INFO
+	//stream.SetLevelInfo(stream.DEBUG)
+
+	fmt.Println("Getting started with Streaming client for RabbitMQ")
+	fmt.Println("Connecting to RabbitMQ streaming ...")
+
+	env, err := stream.NewEnvironment(
+		stream.NewEnvironmentOptions().
+			SetHost("localhost").
+			SetPort(5552).
+			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 := "test"
+	err = env.DeclareStream(streamName,
+		&stream.StreamOptions{
+			MaxLengthBytes: stream.ByteCapacity{}.GB(2),
+		},
+	)
+
+	rProducer, err := ha.NewHAProducer(env, streamName, "producer-ha")
+	CheckErr(err)
+
+	chPublishConfirm := rProducer.NotifyPublishConfirmation()
+	handlePublishConfirm(chPublishConfirm)
+	time.Sleep(4 * time.Second)
+	for i := 0; i < 1000000; i++ {
+		err := rProducer.BatchPublish(CreateArrayMessagesForTesting(10))
+		time.Sleep(10 * time.Millisecond)
+		if i%1000 == 0 {
+			fmt.Println("sent.. " + strconv.Itoa(i))
+		}
+		CheckErr(err)
+	}
+
+	fmt.Println("Press any key to start consuming ")
+	_, _ = reader.ReadString('\n')
+
+	handleMessages := func(consumerContext stream.ConsumerContext, message *amqp.Message) {
+		fmt.Printf("messages consumed: %s \n ", message.Data)
+	}
+
+	consumer, err := env.NewConsumer(context.TODO(), streamName,
+		handleMessages,
+		stream.NewConsumerOptions().
+			SetConsumerName("my_consumer"))
+	CheckErr(err)
+
+	fmt.Println("Press any key to stop ")
+	_, _ = reader.ReadString('\n')
+	time.Sleep(200 * time.Millisecond)
+	err = rProducer.Close()
+	CheckErr(err)
+	err = consumer.Close()
+	CheckErr(err)
+	err = env.DeleteStream(streamName)
+	CheckErr(err)
+	err = env.Close()
+	CheckErr(err)
+}