Merge pull request #33 from heetch/sendmethod

Send method

Author: asdine

codec/encoder.go

@@ -1,30 +1,33 @@
 package codec
 
-import (
-	"github.com/Shopify/sarama"
-)
+// Encoder is a simple interface for any type that can be encoded as an array of bytes
+// in order to be sent as the key or value of a Kafka message. It matches the sarama.Encoder interface.
+type Encoder interface {
+	Encode() ([]byte, error)
+	Length() int
+}
 
-// StringEncoder creates a sarama.Encoder that encodes using the String codec.
-func StringEncoder(v string) sarama.Encoder {
+// StringEncoder creates a Encoder that encodes using the String codec.
+func StringEncoder(v string) Encoder {
 	return &encoder{codec: String(), v: v}
 }
 
-// Int64Encoder creates a sarama.Encoder that encodes using the Int64 codec.
-func Int64Encoder(v int) sarama.Encoder {
+// Int64Encoder creates a Encoder that encodes using the Int64 codec.
+func Int64Encoder(v int) Encoder {
 	return &encoder{codec: Int64(), v: int64(v)}
 }
 
-// Float64Encoder creates a sarama.Encoder that encodes using the Float64 Codec.
-func Float64Encoder(v float64) sarama.Encoder {
+// Float64Encoder creates a Encoder that encodes using the Float64 Codec.
+func Float64Encoder(v float64) Encoder {
 	return &encoder{codec: Float64(), v: v}
 }
 
-// JSONEncoder creates a sarama.Encoder that encodes using the JSON Codec.
-func JSONEncoder(v interface{}) sarama.Encoder {
+// JSONEncoder creates a Encoder that encodes using the JSON Codec.
+func JSONEncoder(v interface{}) Encoder {
 	return &encoder{codec: JSON(), v: v}
 }
 
-// encoder implements the sarama.Encoder interface.
+// encoder implements the Encoder interface.
 type encoder struct {
 	codec Codec
 	v     interface{}

producer/doc_test.go

@@ -59,3 +59,18 @@ func ExampleNewFrom() {
 		panic(err)
 	}
 }
+
+func ExampleProducer_Send() {
+	config := producer.NewConfig("some-id", producer.MessageConverterV1())
+
+	p, err := producer.New(config, endpoints...)
+	if err != nil {
+		panic(err)
+	}
+	defer p.Close()
+
+	_, err = p.Send(context.Background(), "some topic", "some body", producer.StrKey("some key"))
+	if err != nil {
+		panic(err)
+	}
+}

producer/message.go

@@ -0,0 +1,108 @@
+package producer
+
+import (
+	"time"
+
+	"github.com/Shopify/sarama"
+	"github.com/heetch/felice/codec"
+	uuid "github.com/satori/go.uuid"
+)
+
+// Message represents a message to be sent via Kafka.
+// Before sending it, the producer will transform this structure into a
+// sarama.ProducerMessage using the registered Converter.
+type Message struct {
+	// The Kafka topic this Message applies to.
+	Topic string
+
+	// If specified, messages with the same key will be sent to the same Kafka partition.
+	Key sarama.Encoder
+
+	// Body of the Kafka message.
+	Body interface{}
+
+	// The time at which this Message was produced.
+	ProducedAt time.Time
+
+	// Partition where this publication was stored.
+	Partition int32
+
+	// Offset where this publication was stored.
+	Offset int64
+
+	// Headers of the message.
+	Headers map[string]string
+
+	// Unique ID of the message. Defaults to an uuid.
+	ID string
+}
+
+// prepare makes sure the message contains a unique ID and
+// the Headers map memory is allocated.
+func (m *Message) prepare() {
+	if m.ID == "" {
+		m.ID = uuid.Must(uuid.NewV4()).String()
+	}
+
+	if m.Headers == nil {
+		m.Headers = make(map[string]string)
+	}
+}
+
+// NewMessage creates a configured message with a generated unique ID.
+func NewMessage(topic string, body interface{}) *Message {
+	return &Message{
+		Topic:   topic,
+		Body:    body,
+		Headers: make(map[string]string),
+		ID:      uuid.Must(uuid.NewV4()).String(),
+	}
+}
+
+// Option is a function type that receives a pointer to a Message and
+// modifies it in place. Options are intended to customize a message
+// before sending it. You can do this either by passing them as
+// parameters to the New function, or by calling them directly against
+// a Message.
+type Option func(*Message)
+
+// Header is an Option that adds a custom header to the message. You
+// may pass as many Header options to New as you wish. If multiple
+// Header's are defined for the same key, the value of the last one
+// past to New will be the value that appears on the Message.
+func Header(k, v string) Option {
+	return func(m *Message) {
+		m.Headers[k] = v
+	}
+}
+
+// Key is an Option that specifies a key for the message. You should
+// only pass this once to the New function, but if you pass it multiple
+// times, the value set by the final one you pass will be what is set
+// on the Message when it is returned by New.
+func Key(key codec.Encoder) Option {
+	return func(m *Message) {
+		m.Key = key
+	}
+}
+
+// StrKey is an Option that specifies a key for the message as a string.
+func StrKey(key string) Option {
+	return func(m *Message) {
+		m.Key = codec.StringEncoder(key)
+	}
+}
+
+// Int64Key is an Option that specifies a key for the message as an integer.
+func Int64Key(key int) Option {
+	return func(m *Message) {
+		m.Key = codec.Int64Encoder(key)
+	}
+}
+
+// Float64Key is an Option that specifies a key for the message as a float.
+func Float64Key(key float64) Option {
+	return func(m *Message) {
+		m.Key = codec.Float64Encoder(key)
+	}
+}

producer/producer.go

@@ -2,12 +2,10 @@ package producer
 
 import (
 	"context"
-	"time"
 
 	"github.com/Shopify/sarama"
 	"github.com/heetch/felice/codec"
 	"github.com/pkg/errors"
-	uuid "github.com/satori/go.uuid"
 )
 
 // Producer sends messages to Kafka.
@@ -55,6 +53,18 @@ func NewFrom(producer sarama.SyncProducer, config Config) (*Producer, error) {
 	return &Producer{SyncProducer: producer, config: config}, nil
 }
 
+// Send creates and sends a message to Kafka synchronously.
+// It returns the message.Message sent to the brokers.
+func (p *Producer) Send(ctx context.Context, topic string, body interface{}, opts ...Option) (*Message, error) {
+	msg := NewMessage(topic, body)
+	for _, opt := range opts {
+		opt(msg)
+	}
+
+	err := p.SendMessage(ctx, msg)
+	return msg, err
+}
+
 // SendMessage sends the given message to Kafka synchronously.
 func (p *Producer) SendMessage(ctx context.Context, msg *Message) error {
 	select {
@@ -80,57 +90,6 @@ func (p *Producer) SendMessage(ctx context.Context, msg *Message) error {
 	return errors.Wrap(err, "producer: failed to send message")
 }
 
-// Message represents a message to be sent via Kafka.
-// Before sending it, the producer will transform this structure into a
-// sarama.ProducerMessage using the registered Converter.
-type Message struct {
-	// The Kafka topic this Message applies to.
-	Topic string
-
-	// If specified, messages with the same key will be sent to the same Kafka partition.
-	Key sarama.Encoder
-
-	// Body of the Kafka message.
-	Body interface{}
-
-	// The time at which this Message was produced.
-	ProducedAt time.Time
-
-	// Partition where this publication was stored.
-	Partition int32
-
-	// Offset where this publication was stored.
-	Offset int64
-
-	// Headers of the message.
-	Headers map[string]string
-
-	// Unique ID of the message. Defaults to an uuid.
-	ID string
-}
-
-// prepare makes sure the message contains a unique ID and
-// the Headers map memory is allocated.
-func (m *Message) prepare() {
-	if m.ID == "" {
-		m.ID = uuid.Must(uuid.NewV4()).String()
-	}
-
-	if m.Headers == nil {
-		m.Headers = make(map[string]string)
-	}
-}
-
-// NewMessage creates a configured message with a generated unique ID.
-func NewMessage(topic string, body interface{}) *Message {
-	return &Message{
-		Topic:   topic,
-		Body:    body,
-		Headers: make(map[string]string),
-		ID:      uuid.Must(uuid.NewV4()).String(),
-	}
-}
-
 // A MessageConverter transforms a Message into a sarama.ProducerMessage.
 // The role of the converter is to decouple the conventions defined by users from
 // the producer.

producer/producer_test.go

@@ -38,3 +38,23 @@ func TestSendMessage(t *testing.T) {
 		require.EqualError(t, err, "producer: failed to send message: cannot produce message")
 	}
 }
+
+func TestSend(t *testing.T) {
+	msp := mocks.NewSyncProducer(t, nil)
+	cfg := producer.NewConfig("id", producer.MessageConverterV1())
+	p, err := producer.NewFrom(msp, cfg)
+	require.NoError(t, err)
+
+	msp.ExpectSendMessageAndSucceed()
+	msg, err := p.Send(context.Background(), "topic", "message", producer.Int64Key(10), producer.Header("k", "v"))
+	require.NoError(t, err)
+	key, err := msg.Key.Encode()
+	require.NoError(t, err)
+
+	require.EqualValues(t, "10", key)
+	require.Equal(t, "v", msg.Headers["k"])
+
+	msp.ExpectSendMessageAndFail(fmt.Errorf("cannot produce message"))
+	_, err = p.Send(context.Background(), "topic", "message")
+	require.EqualError(t, err, "producer: failed to send message: cannot produce message")
+}