Topic reader implementation (apache#158)

* Topic reader implementation

* Fixed formatting

* Fixed function name

* Fixed the HasNext()

* Fixed formatting stuff

* Fixed queue draining on reconnection

* Removed space

* Fixed deadlock on start

* Added reader example

examples/reader/reader.go

@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+
+	"github.com/apache/pulsar-client-go/pulsar"
+)
+
+func main() {
+	client, err := pulsar.NewClient(pulsar.ClientOptions{URL: "pulsar://localhost:6650"})
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	defer client.Close()
+
+	reader, err := client.CreateReader(pulsar.ReaderOptions{
+		Topic:          "topic-1",
+		StartMessageID: pulsar.EarliestMessageID(),
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer reader.Close()
+
+	for reader.HasNext() {
+		msg, err := reader.Next(context.Background())
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		fmt.Printf("Received message msgId: %#v -- content: '%s'\n",
+			msg.ID(), string(msg.Payload()))
+	}
+}

pulsar/client_impl.go

@@ -118,8 +118,12 @@ func (c *client) Subscribe(options ConsumerOptions) (Consumer, error) {
 }
 
 func (c *client) CreateReader(options ReaderOptions) (Reader, error) {
-	// TODO: Implement reader
-	return nil, nil
+	reader, err := newReader(c, options)
+	if err != nil {
+		return nil, err
+	}
+	c.handlers.Add(reader)
+	return reader, nil
 }
 
 func (c *client) TopicPartitions(topic string) ([]string, error) {

pulsar/consumer.go

@@ -121,6 +121,9 @@ type ConsumerOptions struct {
 	//  failure or exclusive subscriptions). Attempting to enable it on subscriptions to a non-persistent topics or on a
 	//  shared subscription, will lead to the subscription call throwing a PulsarClientException.
 	ReadCompacted bool
+
+	// Mark the subscription as replicated to keep it in sync across clusters
+	ReplicateSubscriptionState bool
 }
 
 // Consumer is an interface that abstracts behavior of Pulsar's consumer

pulsar/consumer_impl.go

@@ -156,15 +156,19 @@ func internalTopicSubscribe(client *client, options ConsumerOptions, topic strin
 				nackRedeliveryDelay = options.NackRedeliveryDelay
 			}
 			opts := &partitionConsumerOpts{
-				topic:               pt,
-				consumerName:        consumerName,
-				subscription:        options.SubscriptionName,
-				subscriptionType:    options.Type,
-				subscriptionInitPos: options.SubscriptionInitialPosition,
-				partitionIdx:        idx,
-				receiverQueueSize:   receiverQueueSize,
-				nackRedeliveryDelay: nackRedeliveryDelay,
-				metadata:            metadata,
+				topic:                      pt,
+				consumerName:               consumerName,
+				subscription:               options.SubscriptionName,
+				subscriptionType:           options.Type,
+				subscriptionInitPos:        options.SubscriptionInitialPosition,
+				partitionIdx:               idx,
+				receiverQueueSize:          receiverQueueSize,
+				nackRedeliveryDelay:        nackRedeliveryDelay,
+				metadata:                   metadata,
+				replicateSubscriptionState: options.ReplicateSubscriptionState,
+				startMessageID:             nil,
+				subscriptionMode:           durable,
+				readCompacted:              options.ReadCompacted,
 			}
 			cons, err := newPartitionConsumer(consumer, client, opts, messageCh)
 			ch <- ConsumerError{