Added Events() and ProduceChannel() getters (issue confluentinc#5)

Author: edenhill

examples/consumer_channel_example/consumer_channel_example.go

@@ -65,7 +65,7 @@ func main() {
 			fmt.Printf("Caught signal %v: terminating\n", sig)
 			run = false
 
-		case ev := <-c.Events:
+		case ev := <-c.Events():
 			switch e := ev.(type) {
 			case kafka.AssignedPartitions:
 				fmt.Fprintf(os.Stderr, "%% %v\n", e)

examples/go-kafkacat/go-kafkacat.go

@@ -60,7 +60,7 @@ func runProducer(config *kafka.ConfigMap, topic string, partition int32) {
 				fmt.Fprintf(os.Stderr, "%% Delivered %v\n", m)
 			}
 		}
-	}(p.Events)
+	}(p.Events())
 
 	reader := bufio.NewReader(os.Stdin)
 	stdinChan := make(chan string)
@@ -110,7 +110,7 @@ func runProducer(config *kafka.ConfigMap, topic string, partition int32) {
 				msg.Value = ([]byte)(line)
 			}
 
-			p.ProduceChannel <- &msg
+			p.ProduceChannel() <- &msg
 		}
 	}
 
@@ -140,7 +140,7 @@ func runConsumer(config *kafka.ConfigMap, topics []string) {
 			fmt.Fprintf(os.Stderr, "%% Terminating on signal %v\n", sig)
 			run = false
 
-		case ev := <-c.Events:
+		case ev := <-c.Events():
 			switch e := ev.(type) {
 			case kafka.AssignedPartitions:
 				fmt.Fprintf(os.Stderr, "%% %v\n", e)

examples/producer_channel_example/producer_channel_example.go

@@ -47,7 +47,7 @@ func main() {
 
 	go func() {
 	outer:
-		for e := range p.Events {
+		for e := range p.Events() {
 			switch ev := e.(type) {
 			case *kafka.Message:
 				m := ev
@@ -68,7 +68,7 @@ func main() {
 	}()
 
 	value := "Hello Go!"
-	p.ProduceChannel <- &kafka.Message{TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny}, Value: []byte(value)}
+	p.ProduceChannel() <- &kafka.Message{TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny}, Value: []byte(value)}
 
 	// wait for delivery report goroutine to finish
 	_ = <-doneChan

kafka/consumer.go

@@ -33,7 +33,7 @@ type RebalanceCb func(*Consumer, Event) error
 
 // Consumer implements a High-level Apache Kafka Consumer instance
 type Consumer struct {
-	Events             chan Event
+	events             chan Event
 	handle             handle
 	eventsChanEnable   bool
 	readerTermChan     chan bool
@@ -202,6 +202,11 @@ func (c *Consumer) Poll(timeoutMs int) (event Event) {
 	return ev
 }
 
+// Events returns the Events channel (if enabled)
+func (c *Consumer) Events() chan Event {
+	return c.events
+}
+
 // Close Consumer instance.
 // The object is no longer usable after this call.
 func (c *Consumer) Close() (err error) {
@@ -231,12 +236,12 @@ func (c *Consumer) Close() (err error) {
 // NewConsumer creates a new high-level Consumer instance.
 //
 // Supported special configuration properties:
-//   go.application.rebalance.enable (bool, false) - Forward rebalancing responsibility to application via the Events channel.
+//   go.application.rebalance.enable (bool, false) - Forward rebalancing responsibility to application via the Events() channel.
 //                                        If set to true the app must handle the AssignedPartitions and
 //                                        RevokedPartitions events and call Assign() and Unassign()
 //                                        respectively.
-//   go.events.channel.enable (bool, false) - Enable the Events channel. Messages and events will be pushed on the Events channel and the Poll() interface will be disabled. (Experimental)
-//   go.events.channel.size (int, 1000) - Events channel size
+//   go.events.channel.enable (bool, false) - Enable the Events() channel. Messages and events will be pushed on the Events() channel and the Poll() interface will be disabled. (Experimental)
+//   go.events.channel.size (int, 1000) - Events() channel size
 //
 // WARNING: Due to the buffering nature of channels (and queues in general) the
 // use of the events channel risks receiving outdated events and
@@ -299,10 +304,10 @@ func NewConsumer(conf *ConfigMap) (*Consumer, error) {
 	}
 
 	if c.eventsChanEnable {
-		c.Events = make(chan Event, eventsChanSize)
+		c.events = make(chan Event, eventsChanSize)
 		c.readerTermChan = make(chan bool)
 
-		/* Start rdkafka consumer queue reader -> Events writer goroutine */
+		/* Start rdkafka consumer queue reader -> events writer goroutine */
 		go consumerReader(c, c.readerTermChan)
 	}
 
@@ -332,7 +337,7 @@ out:
 		case _ = <-termChan:
 			break out
 		default:
-			_, term := c.handle.eventPoll(c.Events, 100, 1000, termChan)
+			_, term := c.handle.eventPoll(c.events, 100, 1000, termChan)
 			if term {
 				break out
 			}

kafka/consumer_performance_test.go

@@ -102,7 +102,7 @@ func handleEvent(c *Consumer, rd *ratedisp, expCnt int, ev Event) bool {
 
 // consume messages through the Events channel
 func eventChannelConsumer(c *Consumer, rd *ratedisp, expCnt int) {
-	for ev := range c.Events {
+	for ev := range c.Events() {
 		if !handleEvent(c, rd, expCnt, ev) {
 			break
 		}
@@ -148,7 +148,7 @@ func testconsumerInit(b *testing.B) int {
 		producerPerfTest(b, "Priming producer", msgcnt, false, false,
 			true,
 			func(p *Producer, m *Message, drChan chan Event) {
-				p.ProduceChannel <- m
+				p.ProduceChannel() <- m
 			})
 	}
 

kafka/producer.go

@@ -38,8 +38,8 @@ import "C"
 
 // Producer implements a High-level Apache Kafka Producer instance
 type Producer struct {
-	Events         chan Event
-	ProduceChannel chan *Message
+	events         chan Event
+	produceChannel chan *Message
 	handle         handle
 
 	// Terminates the poller() goroutine
@@ -114,7 +114,7 @@ func (p *Producer) produce(msg *Message, msgFlags int, deliveryChan chan Event)
 // This is an asynchronous call that enqueues the message on the internal
 // transmit queue, thus returning immediately.
 // The delivery report will be sent on the provided deliveryChan if specified,
-// or on the Producer object's Events channel if not.
+// or on the Producer object's Events() channel if not.
 // Returns an error if message could not be enqueued.
 func (p *Producer) Produce(msg *Message, deliveryChan chan Event) error {
 	return p.produce(msg, 0, deliveryChan)
@@ -140,11 +140,21 @@ func (p *Producer) produceBatch(topic string, msgs []*Message, msgFlags int) err
 	return nil
 }
 
+// Events returns the Events channel (read)
+func (p *Producer) Events() chan Event {
+	return p.events
+}
+
+// ProduceChannel returns the produce *Message channel (write)
+func (p *Producer) ProduceChannel() chan *Message {
+	return p.produceChannel
+}
+
 // Len returns the number of messages and requests waiting to be transmitted to the broker
 // as well as delivery reports queued for the application.
 // Includes messages on ProduceChannel.
 func (p *Producer) Len() int {
-	return len(p.ProduceChannel) + len(p.Events) + int(C.rd_kafka_outq_len(p.handle.rk))
+	return len(p.produceChannel) + len(p.events) + int(C.rd_kafka_outq_len(p.handle.rk))
 }
 
 // Flush and wait for outstanding messages and requests to complete delivery.
@@ -162,7 +172,7 @@ func (p *Producer) Flush(timeoutMs int) int {
 			return p.Len()
 		}
 
-		p.handle.eventPoll(p.Events,
+		p.handle.eventPoll(p.events,
 			int(math.Min(100, remain*1000)), 1000, termChan)
 	}
 
@@ -175,10 +185,10 @@ func (p *Producer) Close() {
 	// Wait for poller() (signaled by closing pollerTermChan)
 	// and channel_producer() (signaled by closing ProduceChannel)
 	close(p.pollerTermChan)
-	close(p.ProduceChannel)
+	close(p.produceChannel)
 	p.handle.waitTerminated(2)
 
-	close(p.Events)
+	close(p.events)
 
 	p.handle.cleanup()
 
@@ -197,8 +207,8 @@ func (p *Producer) Close() {
 //   go.batch.producer (bool, false) - Enable batch producer (experimental for increased performance).
 //                                     These batches do not relate to Kafka message batches in any way.
 //   go.delivery.reports (bool, true) - Forward per-message delivery reports to the
-//                                      Events channel.
-//   go.produce.channel.size (int, 1000000) - ProduceChannel buffer size (in number of messages)
+//                                      Events() channel.
+//   go.produce.channel.size (int, 1000000) - ProduceChannel() buffer size (in number of messages)
 //
 func NewProducer(conf *ConfigMap) (*Producer, error) {
 	p := &Producer{}
@@ -241,8 +251,8 @@ func NewProducer(conf *ConfigMap) (*Producer, error) {
 	p.handle.p = p
 	p.handle.setup()
 	p.handle.rkq = C.rd_kafka_queue_get_main(p.handle.rk)
-	p.Events = make(chan Event, 1000000)
-	p.ProduceChannel = make(chan *Message, produceChannelSize)
+	p.events = make(chan Event, 1000000)
+	p.produceChannel = make(chan *Message, produceChannelSize)
 	p.pollerTermChan = make(chan bool)
 
 	go poller(p, p.pollerTermChan)
@@ -260,11 +270,11 @@ func NewProducer(conf *ConfigMap) (*Producer, error) {
 // channel_producer serves the ProduceChannel channel
 func channelProducer(p *Producer) {
 
-	for m := range p.ProduceChannel {
+	for m := range p.produceChannel {
 		err := p.produce(m, C.RD_KAFKA_MSG_F_BLOCK, nil)
 		if err != nil {
 			m.TopicPartition.Error = err
-			p.Events <- m
+			p.events <- m
 		}
 	}
 
@@ -280,14 +290,14 @@ func channelBatchProducer(p *Producer) {
 	totMsgCnt := 0
 	totBatchCnt := 0
 
-	for m := range p.ProduceChannel {
+	for m := range p.produceChannel {
 		buffered[*m.TopicPartition.Topic] = append(buffered[*m.TopicPartition.Topic], m)
 		bufferedCnt++
 
 	loop2:
 		for true {
 			select {
-			case m, ok := <-p.ProduceChannel:
+			case m, ok := <-p.produceChannel:
 				if !ok {
 					break loop2
 				}
@@ -315,7 +325,7 @@ func channelBatchProducer(p *Producer) {
 			if err != nil {
 				for _, m = range buffered2 {
 					m.TopicPartition.Error = err
-					p.Events <- m
+					p.events <- m
 				}
 			}
 		}
@@ -335,7 +345,7 @@ out:
 			break out
 
 		default:
-			_, term := p.handle.eventPoll(p.Events, 100, 1000, termChan)
+			_, term := p.handle.eventPoll(p.events, 100, 1000, termChan)
 			if term {
 				break out
 			}

kafka/producer_performance_test.go

@@ -90,8 +90,8 @@ func producerPerfTest(b *testing.B, testname string, msgcnt int, withDr bool, ba
 
 	if withDr {
 		doneChan = make(chan int64)
-		drChan = p.Events
-		go deliveryHandler(b, int64(msgcnt), p.Events, doneChan)
+		drChan = p.Events()
+		go deliveryHandler(b, int64(msgcnt), p.Events(), doneChan)
 	}
 
 	if !silent {
@@ -171,15 +171,15 @@ func BenchmarkProducerChannel(b *testing.B) {
 	producerPerfTest(b, "Channel producer (without DR)",
 		0, false, false, false,
 		func(p *Producer, m *Message, drChan chan Event) {
-			p.ProduceChannel <- m
+			p.ProduceChannel() <- m
 		})
 }
 
 func BenchmarkProducerChannelDR(b *testing.B) {
 	producerPerfTest(b, "Channel producer (with DR)",
 		testconf.PerfMsgCount, true, false, false,
 		func(p *Producer, m *Message, drChan chan Event) {
-			p.ProduceChannel <- m
+			p.ProduceChannel() <- m
 		})
 
 }
@@ -188,15 +188,15 @@ func BenchmarkProducerBatchChannel(b *testing.B) {
 	producerPerfTest(b, "Channel producer (without DR, batch channel)",
 		0, false, true, false,
 		func(p *Producer, m *Message, drChan chan Event) {
-			p.ProduceChannel <- m
+			p.ProduceChannel() <- m
 		})
 }
 
 func BenchmarkProducerBatchChannelDR(b *testing.B) {
 	producerPerfTest(b, "Channel producer (DR, batch channel)",
 		0, true, true, false,
 		func(p *Producer, m *Message, drChan chan Event) {
-			p.ProduceChannel <- m
+			p.ProduceChannel() <- m
 		})
 }
 

kafka/producer_test.go

@@ -56,7 +56,7 @@ func TestProducerAPIs(t *testing.T) {
 	// Produce through ProducerChannel, uses default DR channel (Events),
 	// pass Opaque object.
 	myOpq := "My opaque"
-	p.ProduceChannel <- &Message{TopicPartition: TopicPartition{Topic: &topic2, Partition: 0},
+	p.ProduceChannel() <- &Message{TopicPartition: TopicPartition{Topic: &topic2, Partition: 0},
 		Opaque: &myOpq,
 		Value:  []byte("ProducerChannel"), Key: []byte("This is my key")}
 
@@ -85,7 +85,7 @@ func TestProducerAPIs(t *testing.T) {
 
 	// Events chan (2 messages and possibly events)
 	for msgCnt := 0; msgCnt < 2; {
-		ev = <-p.Events
+		ev = <-p.Events()
 		switch e := ev.(type) {
 		case *Message:
 			msgCnt++

kafkatest/go_verifiable_producer/go_verifiable_producer.go

@@ -151,7 +151,7 @@ func runProducer(config *kafka.ConfigMap, topic string) {
 			fmt.Fprintf(os.Stderr, "%% Terminating on signal %v\n", sig)
 			run = false
 
-		case ev := <-p.Events:
+		case ev := <-p.Events():
 			switch e := ev.(type) {
 			case *kafka.Message:
 				run = handleDr(e)