Lastconsumed when no offset (#122)

* Return offset 0 when no offset
Co-authored-by: Gabriele Santomaggio <G.santomaggio@gmail.com>

Author: MarcialRosales

pkg/stream/client.go

@@ -672,6 +672,27 @@ func (c *Client) DeclareStream(streamName string, options *StreamOptions) error
 
 }
 
+func (c *Client) queryOffset(consumerName string, streamName string) (int64, error) {
+	length := 2 + 2 + 4 + 2 + len(consumerName) + 2 + len(streamName)
+
+	resp := c.coordinator.NewResponse(CommandQueryOffset)
+	correlationId := resp.correlationid
+	var b = bytes.NewBuffer(make([]byte, 0, length+4))
+	writeProtocolHeader(b, length, CommandQueryOffset,
+		correlationId)
+
+	writeString(b, consumerName)
+	writeString(b, streamName)
+	err := c.handleWriteWithResponse(b.Bytes(), resp, false)
+	offset := <-resp.data
+	_ = c.coordinator.RemoveResponseById(resp.correlationid)
+	if err.Err != nil {
+		return 0, err.Err
+	}
+
+	return offset.(int64), nil
+}
+
 func (c *Client) DeclareSubscriber(streamName string,
 	messagesHandler MessagesHandler,
 	options *ConsumerOptions) (*Consumer, error) {
@@ -691,32 +712,34 @@ func (c *Client) DeclareSubscriber(streamName string,
 		return nil, fmt.Errorf("message count before storage must be bigger than one")
 	}
 
+	if options.Offset.isLastConsumed() {
+		lastOffset, err := c.queryOffset(options.ConsumerName, streamName)
+		switch err {
+		case nil, OffsetNotFoundError:
+			if err == OffsetNotFoundError {
+				options.Offset.typeOfs = typeFirst
+				options.Offset.offset = 0
+				break
+			} else {
+				options.Offset.offset = lastOffset
+				options.Offset.typeOfs = typeOffset
+				break
+			}
+		default:
+			return nil, err
+		}
+	}
+
 	options.client = c
 	options.streamName = streamName
 	consumer := c.coordinator.NewConsumer(messagesHandler, options)
+
 	length := 2 + 2 + 4 + 1 + 2 + len(streamName) + 2 + 2
 	if options.Offset.isOffset() ||
 		options.Offset.isTimestamp() {
 		length += 8
 	}
 
-	if options.Offset.isLastConsumed() {
-		lastOffset, err := consumer.QueryOffset()
-		if err != nil {
-			_ = c.coordinator.RemoveConsumerById(consumer.ID, Event{
-				Command:    CommandQueryOffset,
-				StreamName: streamName,
-				Name:       consumer.GetName(),
-				Reason:     "error QueryOffset",
-				Err:        err,
-			})
-			return nil, err
-		}
-		options.Offset.offset = lastOffset
-		// here we change the type since typeLastConsumed is not part of the protocol
-		options.Offset.typeOfs = typeOffset
-	}
-
 	// copy the option offset to the consumer offset
 	// the option.offset won't change ( in case we need to retrive the original configuration)
 	// consumer.current offset will be moved when reading

pkg/stream/constants.go

@@ -69,6 +69,7 @@ const (
 	responseCodeAccessRefused                 = uint16(16)
 	responseCodePreconditionFailed            = uint16(17)
 	responseCodePublisherDoesNotExist         = uint16(18)
+	responseCodeNoOffset                      = uint16(19)
 
 	/// responses out of protocol
 	closeChannel         = uint16(60)
@@ -111,6 +112,7 @@ var StreamAlreadyExists = errors.New("Stream Already Exists")
 var VirtualHostAccessFailure = errors.New("Virtual Host Access Failure")
 var SubscriptionIdDoesNotExist = errors.New("Subscription Id Does Not Exist")
 var PublisherDoesNotExist = errors.New("Publisher Does Not Exist")
+var OffsetNotFoundError = errors.New("Offset not found")
 var FrameTooLarge = errors.New("Frame Too Large, the buffer is too big")
 var CodeAccessRefused = errors.New("Resources Access Refused")
 var ConnectionClosed = errors.New("Can't send the message, connection closed")
@@ -137,6 +139,8 @@ func lookErrorCode(errorCode uint16) error {
 		return SubscriptionIdDoesNotExist
 	case responseCodePublisherDoesNotExist:
 		return PublisherDoesNotExist
+	case responseCodeNoOffset:
+		return OffsetNotFoundError
 	case responseCodePreconditionFailed:
 		return PreconditionFailed
 	case responseCodeFrameTooLarge:

pkg/stream/consumer.go

@@ -275,27 +275,7 @@ func (consumer *Consumer) internalStoreOffset() error {
 }
 
 func (consumer *Consumer) QueryOffset() (int64, error) {
-	length := 2 + 2 + 4 + 2 + len(consumer.options.ConsumerName) + 2 + len(consumer.options.streamName)
-
-	resp := consumer.options.client.coordinator.NewResponse(CommandQueryOffset)
-	correlationId := resp.correlationid
-	var b = bytes.NewBuffer(make([]byte, 0, length+4))
-	writeProtocolHeader(b, length, CommandQueryOffset,
-		correlationId)
-
-	writeString(b, consumer.options.ConsumerName)
-	writeString(b, consumer.options.streamName)
-	err := consumer.options.client.handleWriteWithResponse(b.Bytes(), resp, false)
-	if err.Err != nil {
-		return 0, err.Err
-
-	}
-
-	offset := <-resp.data
-	_ = consumer.options.client.coordinator.RemoveResponseById(resp.correlationid)
-
-	return offset.(int64), nil
-
+	return consumer.options.client.queryOffset(consumer.options.ConsumerName, consumer.options.streamName)
 }
 
 /*

pkg/stream/consumer_test.go

@@ -276,6 +276,16 @@ var _ = Describe("Streaming Consumers", func() {
 		Expect(consumer.Close()).NotTo(HaveOccurred())
 	})
 
+	It("last consumed message not raise an error fist time", func() {
+
+		_, err := env.NewConsumer(streamName,
+			func(consumerContext ConsumerContext, message *amqp.Message) {
+			}, NewConsumerOptions().
+				SetOffset(OffsetSpecification{}.LastConsumed()).
+				SetConsumerName("consumer_test"))
+		Expect(err).NotTo(HaveOccurred())
+	})
+
 	It("Subscribe/Unsubscribe count messages manual store", func() {
 		producer, err := env.NewProducer(streamName, nil)
 		Expect(err).NotTo(HaveOccurred())