new repo changes (#31)

* new repo changes

Author: Gsantomaggio

Makefile

@@ -25,10 +25,10 @@ build: vet build-perfTest
 test: vet
 	go test -v  ./pkg/streaming -race -coverprofile=coverage.txt -covermode=atomic
 docker-build: build
-	docker build -t gsantomaggio/go-stream-client:$(VERSION)  .
+	docker build -t pivotalrabbitmq/go-stream-perf-test:$(VERSION)  .
 
 docker-push: docker-build
-	docker push gsantomaggio/go-stream-client:$(VERSION)
+	docker push pivotalrabbitmq/go-stream-perf-test:$(VERSION)
 
 run-perTest: build-perfTest
 	go run perfTest/perftest.go $(PERFTEST_FLAGS)

README.md

@@ -8,7 +8,7 @@ Experimental client for [RabbitMQ Stream Queues](https://github.com/rabbitmq/rab
 ### Download
 ---
 ```
-go get -u github.com/gsantomaggio/go-stream-client@v0.2-alpha
+go get -u github.com/gsantomaggio/rabbitmq-stream-go-client@v0.3-alpha
 ```
 
 ### How to test
@@ -28,7 +28,7 @@ go get -u github.com/gsantomaggio/go-stream-client@v0.2-alpha
 ---
 The performance tool is work in progress, you can use it with docker
 ```
-docker run --network host -it gsantomaggio/go-stream-client silent 
+docker run --network host -it pivotalrabbitmq/go-stream-perf-test silent 
 ```
 
 or directly 

VERSION

@@ -1 +1 @@
-0.2-alpha
+0.3-alpha

perfTest/cmd/silent.go

@@ -51,7 +51,7 @@ func printStats() {
 }
 
 func startSimulation() error {
-	streaming.INFO("Silent Simulation, url: %s producers: %d consumers: %d streams: %s\n", rabbitmqBrokerUrl, producers, consumers, streams)
+	streaming.INFO("Silent (%s) Simulation, url: %s producers: %d consumers: %d streams: %s ", streaming.Version, rabbitmqBrokerUrl, producers, consumers, streams)
 
 	err := initStreams()
 	err = startConsumers()
@@ -149,7 +149,10 @@ func startConsumers() error {
 				Name(uuid.New().String()).
 				MessagesHandler(func(Context streaming.ConsumerContext, message *amqp.Message) {
 					if atomic.AddInt32(&consumerMessageCount, 1)%500 == 0 {
-						_ = Context.Consumer.Commit()
+						err := Context.Consumer.Commit()
+						if err != nil {
+							streaming.ERROR("Error Commit: %s", err)
+						}
 					}
 				}).Build()
 			if err != nil {

perfTest/perftest.go

@@ -2,12 +2,13 @@ package main
 
 import (
 	"github.com/gsantomaggio/go-stream-client/perfTest/cmd"
-	"time"
+	"sync"
 )
 
+var wg sync.WaitGroup
 func main() {
-	cmd.Execute()
-	for true {
-		time.Sleep(1 * time.Second)
-	}
+
+	wg.Add(1)
+	go cmd.Execute()
+	wg.Wait()
 }

pkg/streaming/constants.go

@@ -64,7 +64,7 @@ const (
 	DefaultReadSocketBuffer = 4096 * 2
 
 //
-	Version = "0.2-alpha"
+	Version = "0.3-alpha"
 )
 
 func LookErrorCode(errorCode uint16) string {

pkg/streaming/consumer_creator.go

@@ -14,6 +14,7 @@ type Consumer struct {
 	parameters *ConsumerCreator
 	mutex      *sync.RWMutex
 }
+
 func (consumer *Consumer) GetStream() string {
 	return consumer.parameters.streamName
 }
@@ -133,6 +134,9 @@ func (c *ConsumerCreator) Build() (*Consumer, error) {
 					return
 				}
 
+			case data := <-consumer.response.data:
+				consumer.setOffset(data.(int64))
+
 			case messages := <-consumer.response.messages:
 				for _, message := range messages {
 					c.messagesHandler(ConsumerContext{Consumer: consumer}, message)

pkg/streaming/response.go

@@ -191,7 +191,11 @@ func (c *Client) handleDeliver(r *bufio.Reader) {
 	consumer, _ := c.coordinator.GetConsumerById(subscriptionId)
 
 	_ = ReadByte(r)
-	_ = ReadByte(r)
+	chunkType := ReadByte(r)
+	if chunkType != 0 {
+		WARN("Invalid chunkType: %d ", chunkType)
+	}
+
 	_ = ReadUShort(r)
 	numRecords, _ := ReadUInt(r)
 	_ = ReadInt64(r) // timestamp
@@ -210,7 +214,7 @@ func (c *Client) handleDeliver(r *bufio.Reader) {
 	var offsetLimit int64 = -1
 
 	if consumer.parameters.offsetSpecification.isOffset() {
-		offsetLimit = consumer.parameters.offsetSpecification.offset
+		offsetLimit = consumer.getOffset()
 	}
 	//if
 
@@ -235,14 +239,15 @@ func (c *Client) handleDeliver(r *bufio.Reader) {
 				batchConsumingMessages = append(batchConsumingMessages, msg)
 			}
 
+		} else {
+			WARN("entryType Not Handled %d", entryType)
 		}
 		numRecords--
 		offset++
-		consumer.setOffset(offset)
 	}
 
 
-	//consumer.response.code <- Code{id: ResponseCodeOk}
+	consumer.response.data <- offset
 	consumer.response.messages <- batchConsumingMessages
 
 }