Implement AckSuccesses and SimpleProducer

AckSuccesses causes a nil error for every successful message, making it possible
to be fake synchrony by listening once after every send (assuming no buffering).
Wrap this logic in a SimpleProducer, and migrate the old Producer test to be
a SimpleProducer test, since the API is closer.

producer.go

@@ -13,7 +13,8 @@ type ProducerConfig struct {
 	Compression         CompressionCodec // The type of compression to use on messages (defaults to no compression).
 	MaxBufferedMessages uint             // The maximum number of messages permitted to buffer before flushing.
 	MaxBufferedBytes    uint             // The maximum number of message bytes permitted to buffer before flushing.
-	MaxBufferTime       time.Duration    // The maximum amount of time permitted to buffer before flushing.
+	MaxBufferTime       time.Duration    // The maximum amount of time permitted to buffer before flushing (or zero for no timer).
+	AckSuccesses        bool             // When true, every successful delivery causes a nil to be sent on the Errors channel.
 }
 
 // Producer publishes Kafka messages. It routes messages to the correct broker, refreshing metadata as appropriate,
@@ -197,9 +198,14 @@ func (d *dispatcher) getQueue(msg *pendingMessage) *msgQueue {
 	return d.queues[msg.topic][msg.partition]
 }
 
+func (d *dispatcher) cleanup() {
+	for _, batcher := range d.batchers {
+		close(batcher.msgs)
+	}
+}
+
 func (d *dispatcher) dispatch() {
 	for msg := range d.msgs {
-		Logger.Println("dispatcher", msg)
 
 		queue := d.getQueue(msg)
 
@@ -220,7 +226,6 @@ func (d *dispatcher) dispatch() {
 				if d.batchers[queue.broker] == nil {
 					d.createBatcher(queue.broker)
 				}
-				Logger.Println("dispatching")
 				d.batchers[queue.broker].msgs <- msg
 			} else {
 				queue.backlog = append(queue.backlog, msg)
@@ -267,6 +272,7 @@ func (d *dispatcher) dispatch() {
 			}
 		}
 	}
+	d.cleanup()
 }
 
 func (b *batcher) buildRequest() *ProduceRequest {
@@ -341,13 +347,37 @@ func (b *batcher) flush() {
 		return
 	}
 
-	Logger.Println("flushing")
-
 	response, err := b.broker.Produce(b.prod.client.id, request)
 
 	switch err {
 	case nil:
-		break
+		if response != nil {
+			for topic, tmp := range b.buffers {
+				for part, buffer := range tmp {
+					block := response.GetBlock(topic, part)
+
+					if block == nil {
+						b.handleError(buffer, IncompleteResponse)
+					} else {
+						switch block.Err {
+						case NoError:
+							if b.prod.config.AckSuccesses {
+								for _ = range buffer {
+									b.prod.errors <- nil
+								}
+							}
+						case UnknownTopicOrPartition, NotLeaderForPartition, LeaderNotAvailable:
+							b.handleError(buffer, err)
+						default:
+							for _, msg := range buffer {
+								b.prod.errors <- &ProduceError{msg.topic, msg.origKey, msg.origValue, err}
+							}
+						}
+
+					}
+				}
+			}
+		}
 	case EncodingError:
 		for _, tmp := range b.buffers {
 			for _, buffer := range tmp {
@@ -364,30 +394,6 @@ func (b *batcher) flush() {
 		}
 	}
 
-	if response != nil {
-		for topic, tmp := range b.buffers {
-			for part, buffer := range tmp {
-				block := response.GetBlock(topic, part)
-
-				if block == nil {
-					b.handleError(buffer, IncompleteResponse)
-				} else {
-					switch block.Err {
-					case NoError:
-						break
-					case UnknownTopicOrPartition, NotLeaderForPartition, LeaderNotAvailable:
-						b.handleError(buffer, err)
-					default:
-						for _, msg := range buffer {
-							b.prod.errors <- &ProduceError{msg.topic, msg.origKey, msg.origValue, err}
-						}
-					}
-
-				}
-			}
-		}
-	}
-
 	b.buffers = make(map[string]map[int32][]*pendingMessage)
 	b.timer.Reset()
 }
@@ -396,11 +402,9 @@ func (b *batcher) processMessages() {
 	for {
 		select {
 		case msg := <-b.msgs:
-			Logger.Println("batcher", msg)
 
 			if msg == nil {
 				b.flush()
-				Logger.Println("batcher exiting")
 				return
 			}
 

simple_producer.go

@@ -0,0 +1,49 @@
+package sarama
+
+// SimpleProducer publishes Kafka messages. It routes messages to the correct broker, refreshing metadata as appropriate,
+// and parses responses for errors. You must call Close() on a producer to avoid leaks, it may not be garbage-collected automatically when
+// it passes out of scope (this is in addition to calling Close on the underlying client, which is still necessary).
+type SimpleProducer struct {
+	producer *Producer
+	topic    string
+}
+
+// NewSimpleProducer creates a new SimpleProducer using the given client and topic.
+func NewSimpleProducer(client *Client, topic string) (*SimpleProducer, error) {
+	if topic == "" {
+		return nil, ConfigurationError("Empty topic")
+	}
+
+	config := new(ProducerConfig)
+	config.RequiredAcks = 1
+	config.AckSuccesses = true
+
+	prod, err := NewProducer(client, config)
+
+	if err != nil {
+		return nil, err
+	}
+
+	return &SimpleProducer{prod, topic}, nil
+}
+
+// SendMessage produces a message with the given key and value. The partition to send to is selected
+// at randome. To send strings as either key or value, see the StringEncoder type.
+func (sp *SimpleProducer) SendMessage(key, value Encoder) error {
+	sp.producer.SendMessage(sp.topic, key, value)
+
+	err := <-sp.producer.Errors() // we always get something because AckSuccesses is true
+
+	if err != nil {
+		return err.Err
+	}
+
+	return nil
+}
+
+// Close shuts down the producer and flushes any messages it may have buffered. You must call this function before
+// a producer object passes out of scope, as it may otherwise leak memory. You must call this before calling Close
+// on the underlying client.
+func (sp *SimpleProducer) Close() error {
+	return sp.producer.Close()
+}

producer_test.go → simple_producer_test.go

@@ -51,7 +51,7 @@ func TestSimpleProducer(t *testing.T) {
 	}
 	defer client.Close()
 
-	producer, err := NewProducer(client, "my_topic", &ProducerConfig{RequiredAcks: WaitForLocal})
+	producer, err := NewSimpleProducer(client, "my_topic")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -65,7 +65,7 @@ func TestSimpleProducer(t *testing.T) {
 	}
 }
 
-func ExampleProducer() {
+func ExampleSimpleProducer() {
 	client, err := NewClient("my_client", []string{"localhost:9092"}, nil)
 	if err != nil {
 		panic(err)
@@ -74,7 +74,7 @@ func ExampleProducer() {
 	}
 	defer client.Close()
 
-	producer, err := NewProducer(client, "my_topic", &ProducerConfig{RequiredAcks: WaitForLocal})
+	producer, err := NewSimpleProducer(client, "my_topic")
 	if err != nil {
 		panic(err)
 	}