Decoder takes topic as an arg.

cswank · Jan 22, 2019 · 2aef792 · 2aef792
1 parent 977ecfe
commit 2aef792
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 4 deletions.
diff --git a/examples/plugin/protobuf.go b/examples/plugin/protobuf.go
@@ -13,7 +13,7 @@ import (
 //     kcli -d ./protobuf.so
 type Protobuf struct{}
 
-func (p Protobuf) Decode(b []byte) ([]byte, error) {
+func (p Protobuf) Decode(topic string, b []byte) ([]byte, error) {
 	var x person.Person
 	err := proto.Unmarshal(b, &x)
 	if err != nil {

diff --git a/internal/kafka/kafka.go b/internal/kafka/kafka.go
@@ -14,13 +14,13 @@ import (
 )
 
 type Decoder interface {
-	Decode([]byte) ([]byte, error)
+	Decode(topic string, data []byte) ([]byte, error)
 }
 
 // plainDecoder is the default Decoder
 type plainDecoder struct{}
 
-func (p plainDecoder) Decode(b []byte) ([]byte, error) { return b, nil }
+func (p plainDecoder) Decode(topic string, data []byte) ([]byte, error) { return data, nil }
 
 //Client fetches from kafka
 type Client struct {
@@ -187,7 +187,7 @@ func (c *Client) GetPartition(part Partition, end int, f func([]byte) bool) ([]M
 		select {
 		case msg = <-pc.Messages():
 			if f(msg.Value) {
-				val, err := c.decoder.Decode(msg.Value)
+				val, err := c.decoder.Decode(part.Topic, msg.Value)
 				if err != nil {
 					return nil, err
 				}