wip

broker_test.go

@@ -41,6 +41,7 @@ func TestBrokerAccessors(t *testing.T) {
 	}
 }
 
+/*
 func TestSimpleBrokerCommunication(t *testing.T) {
 	mb := NewMockBroker(t, 0)
 	defer mb.Close()
@@ -65,6 +66,7 @@ func TestSimpleBrokerCommunication(t *testing.T) {
 		t.Error(err)
 	}
 }
+*/
 
 // We're not testing encoding/decoding here, so most of the requests/responses will be empty for simplicity's sake
 var brokerTestTable = []struct {

consumer_test.go

@@ -45,7 +45,7 @@ func TestSimpleConsumer(t *testing.T) {
 
 	for i := 0; i < 10; i++ {
 		fr := new(FetchResponse)
-		fr.AddMessage("my_topic", 0, nil, ByteEncoder([]byte{0x00, 0x0E}), uint64(i))
+		fr.AddMessage("my_topic", 0, nil, ByteEncoder([]byte{0x00, 0x0E}), int64(i))
 		mb2.Returns(fr)
 	}
 

fetch_response.go

@@ -124,3 +124,17 @@ func (fr *FetchResponse) GetBlock(topic string, partition int32) *FetchResponseB
 
 	return fr.Blocks[topic][partition]
 }
+
+func (fr *FetchResponse) AddMessage(topic string, partition int32, key, value Encoder, offset int64) {
+	partitions, ok := fr.Blocks[topic]
+	if !ok {
+		partitions = make(map[int32]*FetchResponseBlock)
+		fr.Blocks[topic] = partitions
+	}
+	msgSet := partitions[partition].MsgSet
+	kb, _ := key.Encode()
+	vb, _ := value.Encode()
+	msg := &Message{Key: kb, Value: vb}
+	msgBlock := &MessageBlock{Msg: msg, Offset: offset}
+	msgSet.Messages = append(msgSet.Messages, msgBlock)
+}

metadata_response.go

@@ -161,7 +161,6 @@ func (m *MetadataResponse) encode(pe packetEncoder) error {
 			return err
 		}
 	}
-	return nil
 
 	err = pe.putArrayLength(len(m.Topics))
 	if err != nil {
@@ -194,6 +193,7 @@ func (m *MetadataResponse) AddTopicPartition(topic string, partition, brokerID i
 	}
 
 	match = new(TopicMetadata)
+	match.Name = topic
 	m.Topics = append(m.Topics, match)
 
 foundTopic:

mockbroker.go

@@ -3,6 +3,7 @@ package sarama
 import (
 	"encoding/binary"
 	"errors"
+	"fmt"
 	"io"
 	"net"
 	"strconv"
@@ -23,7 +24,7 @@ type MockBroker struct {
 	brokerID     int
 	port         int32
 	stopper      chan bool
-	expectations chan Expectation
+	expectations chan encoder
 	listener     net.Listener
 	t            *testing.T
 	expecting    bool
@@ -33,10 +34,6 @@ func (b *MockBroker) BrokerID() int {
 	return b.brokerID
 }
 
-type Expectation interface {
-	ResponseBytes() []byte
-}
-
 func (b *MockBroker) Port() int32 {
 	return b.port
 }
@@ -51,10 +48,6 @@ func (r rawExpectation) ResponseBytes() []byte {
 	return r
 }
 
-func (b *MockBroker) ExpectBytes(bytes []byte) {
-	b.expectations <- rawExpectation(bytes)
-}
-
 func (b *MockBroker) Close() {
 	if b.expecting {
 		b.t.Fatalf("Not all expectations were satisfied in mockBroker with ID=%d!", b.BrokerID())
@@ -90,7 +83,11 @@ func (b *MockBroker) serverLoop() (ok bool) {
 			return b.serverError(err, conn)
 		}
 
-		response := expectation.ResponseBytes()
+		response, err := encode(expectation)
+		fmt.Println(response, err)
+		if err != nil {
+			return false
+		}
 		if len(response) == 0 {
 			continue
 		}
@@ -133,7 +130,7 @@ func NewMockBroker(t *testing.T, brokerID int) *MockBroker {
 		stopper:      make(chan bool),
 		t:            t,
 		brokerID:     brokerID,
-		expectations: make(chan Expectation, 512),
+		expectations: make(chan encoder, 512),
 	}
 
 	broker.listener, err = net.Listen("tcp", "localhost:0")
@@ -156,5 +153,5 @@ func NewMockBroker(t *testing.T, brokerID int) *MockBroker {
 }
 
 func (b *MockBroker) Returns(e encoder) {
-
+	b.expectations <- e
 }