Index: tools/consumer/consumer.go =================================================================== --- tools/consumer/consumer.go (revision 1186859) +++ tools/consumer/consumer.go (working copy) @@ -20,7 +20,6 @@ * of their respective owners. */ - package main import ( @@ -53,7 +52,6 @@ flag.BoolVar(&printmessage, "printmessage", true, "print the message details to stdout") } - func main() { flag.Parse() fmt.Println("Consuming Messages :") @@ -87,7 +85,7 @@ go func() { for { sig := <-signal.Incoming - if sig.(signal.UnixSignal) == syscall.SIGINT { + if sig.(os.UnixSignal) == syscall.SIGINT { quit <- true } } Index: tools/publisher/publisher.go =================================================================== --- tools/publisher/publisher.go (revision 1186859) +++ tools/publisher/publisher.go (working copy) @@ -34,6 +34,7 @@ var partition int var message string var messageFile string +var compress bool func init() { flag.StringVar(&hostname, "hostname", "localhost:9092", "host:port string for the kafka server") @@ -41,6 +42,7 @@ flag.IntVar(&partition, "partition", 0, "partition to publish to") flag.StringVar(&message, "message", "", "message to publish") flag.StringVar(&messageFile, "messagefile", "", "read message from this file") + flag.BoolVar(&compress, "compress", false, "compress the messages published") } func main() { @@ -64,12 +66,24 @@ payload := make([]byte, stat.Size) file.Read(payload) timing := kafka.StartTiming("Sending") - broker.Publish(kafka.NewMessage(payload)) + + if compress { + broker.Publish(kafka.NewCompressedMessage(payload)) + } else { + broker.Publish(kafka.NewMessage(payload)) + } + timing.Print() file.Close() } else { timing := kafka.StartTiming("Sending") - broker.Publish(kafka.NewMessage([]byte(message))) + + if compress { + broker.Publish(kafka.NewCompressedMessage([]byte(message))) + } else { + broker.Publish(kafka.NewMessage([]byte(message))) + } + timing.Print() } } Index: kafka_test.go =================================================================== --- kafka_test.go (revision 1186859) +++ kafka_test.go (working copy) @@ -20,7 +20,6 @@ * of their respective owners. */ - package kafka import ( @@ -28,12 +27,13 @@ //"fmt" "bytes" "container/list" + "compress/gzip" ) func TestMessageCreation(t *testing.T) { payload := []byte("testing") msg := NewMessage(payload) - if msg.magic != 0 { + if msg.magic != 1 { t.Errorf("magic incorrect") t.Fail() } @@ -45,34 +45,144 @@ } } +func TestMagic0MessageEncoding(t *testing.T) { + // generated by kafka-rb: + // test the old message format + expected := []byte{0x00, 0x00, 0x00, 0x0c, 0x00, 0xe8, 0xf3, 0x5a, 0x06, 0x74, 0x65, 0x73, 0x74, 0x69, 0x6e, 0x67} + msgDecoded := Decode(expected, DefaultCodecsMap) + if msgDecoded == nil { + t.Fail() + } + payload := []byte("testing") + if !bytes.Equal(payload, msgDecoded.payload) { + t.Fatal("bytes not equal") + } + chksum := []byte{0xE8, 0xF3, 0x5A, 0x06} + if !bytes.Equal(chksum, msgDecoded.checksum[:]) { + t.Fatal("checksums do not match") + } + if msgDecoded.magic != 0 { + t.Fatal("magic incorrect") + } +} + func TestMessageEncoding(t *testing.T) { + payload := []byte("testing") msg := NewMessage(payload) // generated by kafka-rb: - expected := []byte{0x00, 0x00, 0x00, 0x0c, 0x00, 0xe8, 0xf3, 0x5a, 0x06, 0x74, 0x65, 0x73, 0x74, 0x69, 0x6e, 0x67} + expected := []byte{0x00, 0x00, 0x00, 0x0d, 0x01, 0x00, 0xe8, 0xf3, 0x5a, 0x06, 0x74, 0x65, 0x73, 0x74, 0x69, 0x6e, 0x67} if !bytes.Equal(expected, msg.Encode()) { - t.Fail() + t.Fatalf("expected: % X\n but got: % X", expected, msg.Encode()) } // verify round trip - msgDecoded := Decode(msg.Encode()) - if !bytes.Equal(msgDecoded.payload, payload) { - t.Fail() + msgDecoded := DecodeWithDefaultCodecs(msg.Encode()) + + if msgDecoded == nil { + t.Fatal("message is nil") } if !bytes.Equal(msgDecoded.payload, payload) { - t.Fail() + t.Fatal("bytes not equal") } chksum := []byte{0xE8, 0xF3, 0x5A, 0x06} - if !bytes.Equal(msgDecoded.checksum[:], chksum) { - t.Fail() + if !bytes.Equal(chksum, msgDecoded.checksum[:]) { + t.Fatal("checksums do not match") } - if msgDecoded.magic != 0 { - t.Fail() + if msgDecoded.magic != 1 { + t.Fatal("magic incorrect") } } +func TestCompressedMessageEncoding(t *testing.T) { + payload := []byte("testing") + msg := NewCompressedMessage(payload) + + expectedPayload := []byte{0x1F, 0x8B, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x04, + 0xFF, 0x2A, 0x49, 0x2D, 0x2E, 0xC9, 0xCC, 0x4B, 0x07, + 0x04, 0x00, 0x00, 0xFF, 0xFF, 0x06, 0x5A, 0xF3, 0xE8, + 0x07, 0x00, 0x00, 0x00} + + expectedHeader := []byte{0x00, 0x00, 0x00, 0x25, 0x01, 0x01, 0x90, 0x15, 0x84, 0x2F} + + expected := make([]byte, len(expectedHeader)+len(expectedPayload)) + n := copy(expected, expectedHeader) + copy(expected[n:], expectedPayload) + + if msg.compression != 1 { + t.Fatalf("expected compression: 1 but got: %b", msg.compression) + } + + zipper, _ := gzip.NewReader(bytes.NewBuffer(msg.payload)) + uncompressed := make([]byte, 100) + n, _ = zipper.Read(uncompressed) + uncompressed = uncompressed[:n] + zipper.Close() + + if !bytes.Equal(uncompressed, payload) { + t.Fatalf("uncompressed: % X \npayload: % X bytes not equal", uncompressed, payload) + } + + if !bytes.Equal(expected, msg.Encode()) { + t.Fatalf("expected: % X\n but got: % X", expected, msg.Encode()) + } + + // verify round trip + msgDecoded := Decode(msg.Encode(), DefaultCodecsMap) + + if msgDecoded == nil { + t.Fatal("message is nil") + } + + if !bytes.Equal(msgDecoded.payload, payload) { + t.Fatal("bytes not equal") + } + chksum := []byte{0x90, 0x15, 0x84, 0x2F} + if !bytes.Equal(chksum, msgDecoded.checksum[:]) { + t.Fatal("checksums do not match") + } + if msgDecoded.magic != 1 { + t.Fatal("magic incorrect") + } +} + +func TestLongCompressedMessageRoundTrip(t *testing.T) { + payloadBuf := bytes.NewBuffer([]byte{}) + // make the test bigger than buffer allocated in the Decode + for i := 0; i < 15; i++ { + payloadBuf.Write([]byte("testing123 ")) + } + + payload := payloadBuf.Bytes() + msg := NewCompressedMessage(payload) + + zipper, _ := gzip.NewReader(bytes.NewBuffer(msg.payload)) + uncompressed := make([]byte, 200) + n, _ := zipper.Read(uncompressed) + uncompressed = uncompressed[:n] + zipper.Close() + + if !bytes.Equal(uncompressed, payload) { + t.Fatalf("uncompressed: % X \npayload: % X bytes not equal", uncompressed, payload) + } + + // verify round trip + msgDecoded := Decode(msg.Encode(), DefaultCodecsMap) + + if msgDecoded == nil { + t.Fatal("message is nil") + } + + if !bytes.Equal(msgDecoded.payload, payload) { + t.Fatal("bytes not equal") + } + if msgDecoded.magic != 1 { + t.Fatal("magic incorrect") + } +} + func TestRequestHeaderEncoding(t *testing.T) { broker := newBroker("localhost:9092", "test", 0) request := broker.EncodeRequestHeader(REQUEST_PRODUCE) @@ -88,7 +198,6 @@ } } - func TestPublishRequestEncoding(t *testing.T) { payload := []byte("testing") msg := NewMessage(payload) @@ -99,13 +208,14 @@ request := pubBroker.broker.EncodePublishRequest(messages) // generated by kafka-rb: - expected := []byte{0x00, 0x00, 0x00, 0x20, 0x00, 0x00, 0x00, 0x04, 0x74, 0x65, 0x73, 0x74, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x10, 0x00, 0x00, 0x00, 0x0c, - 0x00, 0xe8, 0xf3, 0x5a, 0x06, 0x74, 0x65, 0x73, 0x74, 0x69, 0x6e, 0x67} + expected := []byte{0x00, 0x00, 0x00, 0x21, 0x00, 0x00, 0x00, 0x04, 0x74, 0x65, 0x73, 0x74, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x11, 0x00, 0x00, 0x00, 0x0d, + /* magic comp ...... chksum .... .. payload .. */ + 0x01, 0x00, 0xe8, 0xf3, 0x5a, 0x06, 0x74, 0x65, 0x73, 0x74, 0x69, 0x6e, 0x67} if !bytes.Equal(expected, request) { t.Errorf("expected length: %d but got: %d", len(expected), len(request)) - t.Errorf("expected: %X\n but got: %X", expected, request) + t.Errorf("expected: % X\n but got: % X", expected, request) t.Fail() } } @@ -122,7 +232,7 @@ if !bytes.Equal(expected, request) { t.Errorf("expected length: %d but got: %d", len(expected), len(request)) - t.Errorf("expected: %X\n but got: %X", expected, request) + t.Errorf("expected: % X\n but got: % X", expected, request) t.Fail() } } Index: src/consumer.go =================================================================== --- src/consumer.go (revision 1186859) +++ src/consumer.go (working copy) @@ -34,6 +34,7 @@ broker *Broker offset uint64 maxSize uint32 + codecs map[byte]PayloadCodec } // Create a new broker consumer @@ -45,7 +46,8 @@ func NewBrokerConsumer(hostname string, topic string, partition int, offset uint64, maxSize uint32) *BrokerConsumer { return &BrokerConsumer{broker: newBroker(hostname, topic, partition), offset: offset, - maxSize: maxSize} + maxSize: maxSize, + codecs: DefaultCodecsMap} } // Simplified consumer that defaults the offset and maxSize to 0. @@ -55,9 +57,18 @@ func NewBrokerOffsetConsumer(hostname string, topic string, partition int) *BrokerConsumer { return &BrokerConsumer{broker: newBroker(hostname, topic, partition), offset: 0, - maxSize: 0} + maxSize: 0, + codecs: DefaultCodecsMap} } +// Add Custom Payload Codecs for Consumer Decoding +// payloadCodecs - an array of PayloadCodec implementations +func (consumer *BrokerConsumer) AddCodecs(payloadCodecs []PayloadCodec) { + // merge to the default map, so one 'could' override the default codecs.. + for k, v := range codecsMap(payloadCodecs) { + consumer.codecs[k] = v, true + } +} func (consumer *BrokerConsumer) ConsumeOnChannel(msgChan chan *Message, pollTimeoutMs int64, quit chan bool) (int, os.Error) { conn, err := consumer.broker.connect() @@ -111,7 +122,6 @@ return num, err } - func (consumer *BrokerConsumer) consumeWithConn(conn *net.TCPConn, handlerFunc MessageHandlerFunc) (int, os.Error) { _, err := conn.Write(consumer.broker.EncodeConsumeRequest(consumer.offset, consumer.maxSize)) if err != nil { @@ -129,7 +139,7 @@ // parse out the messages var currentOffset uint64 = 0 for currentOffset <= uint64(length-4) { - msg := Decode(payload[currentOffset:]) + msg := Decode(payload[currentOffset:], consumer.codecs) if msg == nil { return num, os.NewError("Error Decoding Message") } @@ -145,7 +155,6 @@ return num, err } - // Get a list of valid offsets (up to maxNumOffsets) before the given time, where // time is in milliseconds (-1, from the latest offset available, -2 from the smallest offset available) // The result is a list of offsets, in descending order. Index: src/kafka.go =================================================================== --- src/kafka.go (revision 1186859) +++ src/kafka.go (working copy) @@ -34,11 +34,9 @@ ) const ( - MAGIC_DEFAULT = 0 - NETWORK = "tcp" + NETWORK = "tcp" ) - type Broker struct { topic string partition int @@ -51,7 +49,6 @@ hostname: hostname} } - func (b *Broker) connect() (conn *net.TCPConn, error os.Error) { raddr, err := net.ResolveTCPAddr(NETWORK, b.hostname) if err != nil { Index: src/publisher.go =================================================================== --- src/publisher.go (revision 1186859) +++ src/publisher.go (working copy) @@ -27,7 +27,6 @@ "os" ) - type BrokerPublisher struct { broker *Broker } @@ -36,7 +35,6 @@ return &BrokerPublisher{broker: newBroker(hostname, topic, partition)} } - func (b *BrokerPublisher) Publish(message *Message) (int, os.Error) { messages := list.New() messages.PushBack(message) @@ -50,7 +48,8 @@ } defer conn.Close() // TODO: MULTIPRODUCE - num, err := conn.Write(b.broker.EncodePublishRequest(messages)) + request := b.broker.EncodePublishRequest(messages) + num, err := conn.Write(request) if err != nil { return -1, err } Index: src/converts.go =================================================================== --- src/converts.go (revision 1186859) +++ src/converts.go (working copy) @@ -22,12 +22,10 @@ package kafka - import ( "encoding/binary" ) - func uint16bytes(value int) []byte { result := make([]byte, 2) binary.BigEndian.PutUint16(result, uint16(value)) Index: src/message.go =================================================================== --- src/message.go (revision 1186859) +++ src/message.go (working copy) @@ -20,7 +20,6 @@ * of their respective owners. */ - package kafka import ( @@ -30,9 +29,16 @@ "log" ) +const ( + // Compression Support uses '1' - https://cwiki.apache.org/confluence/display/KAFKA/Compression + MAGIC_DEFAULT = 1 + // magic + compression + chksum + NO_LEN_HEADER_SIZE = 1 + 1 + 4 +) type Message struct { magic byte + compression byte checksum [4]byte payload []byte offset uint64 // only used after decoding @@ -51,26 +57,44 @@ return string(m.payload) } -func NewMessage(payload []byte) *Message { +func NewMessageWithCodec(payload []byte, codec PayloadCodec) *Message { message := &Message{} message.magic = byte(MAGIC_DEFAULT) - binary.BigEndian.PutUint32(message.checksum[0:], crc32.ChecksumIEEE(payload)) - message.payload = payload + message.compression = codec.Id() + message.payload = codec.Encode(payload) + binary.BigEndian.PutUint32(message.checksum[0:], crc32.ChecksumIEEE(message.payload)) return message } -// MESSAGE SET: +// Default is is create a message with no compression +func NewMessage(payload []byte) *Message { + return NewMessageWithCodec(payload, DefaultCodecsMap[NO_COMPRESSION_ID]) +} + +// Create a Message using the default compression method (gzip) +func NewCompressedMessage(payload []byte) *Message { + return NewMessageWithCodec(payload, DefaultCodecsMap[GZIP_COMPRESSION_ID]) +} + +// MESSAGE SET: func (m *Message) Encode() []byte { - msgLen := 1 + 4 + len(m.payload) + msgLen := NO_LEN_HEADER_SIZE + len(m.payload) msg := make([]byte, 4+msgLen) binary.BigEndian.PutUint32(msg[0:], uint32(msgLen)) msg[4] = m.magic - copy(msg[5:], m.checksum[0:]) - copy(msg[9:], m.payload) + msg[5] = m.compression + copy(msg[6:], m.checksum[0:]) + copy(msg[10:], m.payload) + return msg } -func Decode(packet []byte) *Message { +func DecodeWithDefaultCodecs(packet []byte) *Message { + return Decode(packet, DefaultCodecsMap) +} + +func Decode(packet []byte, payloadCodecsMap map[byte]PayloadCodec) *Message { + length := binary.BigEndian.Uint32(packet[0:]) if length > uint32(len(packet[4:])) { log.Printf("length mismatch, expected at least: %X, was: %X\n", length, len(packet[4:])) @@ -79,25 +103,43 @@ msg := Message{} msg.totalLength = length msg.magic = packet[4] - copy(msg.checksum[:], packet[5:9]) - payloadLength := length - 1 - 4 - msg.payload = packet[9 : 9+payloadLength] + rawPayload := []byte{} + if msg.magic == 0 { + msg.compression = byte(0) + copy(msg.checksum[:], packet[5:9]) + payloadLength := length - 1 - 4 + rawPayload = packet[9 : 9+payloadLength] + } else if msg.magic == MAGIC_DEFAULT { + msg.compression = packet[5] + copy(msg.checksum[:], packet[6:10]) + payloadLength := length - NO_LEN_HEADER_SIZE + + rawPayload = packet[10 : 10+payloadLength] + } else { + log.Printf("incorrect magic, expected: %X was: %X\n", MAGIC_DEFAULT, msg.magic) + return nil + } + payloadChecksum := make([]byte, 4) - binary.BigEndian.PutUint32(payloadChecksum, crc32.ChecksumIEEE(msg.payload)) + binary.BigEndian.PutUint32(payloadChecksum, crc32.ChecksumIEEE(rawPayload)) if !bytes.Equal(payloadChecksum, msg.checksum[:]) { - log.Printf("checksum mismatch, expected: %X was: %X\n", payloadChecksum, msg.checksum[:]) + msg.Print() + log.Printf("checksum mismatch, expected: % X was: % X\n", payloadChecksum, msg.checksum[:]) return nil } + msg.payload = payloadCodecsMap[msg.compression].Decode(rawPayload) + return &msg } func (msg *Message) Print() { log.Println("----- Begin Message ------") log.Printf("magic: %X\n", msg.magic) + log.Printf("compression: %X\n", msg.compression) log.Printf("checksum: %X\n", msg.checksum) if len(msg.payload) < 1048576 { // 1 MB - log.Printf("payload: %X\n", msg.payload) + log.Printf("payload: % X\n", msg.payload) log.Printf("payload(string): %s\n", msg.PayloadString()) } else { log.Printf("long payload, length: %d\n", len(msg.payload)) Index: src/payload_codec.go =================================================================== --- src/payload_codec.go (revision 0) +++ src/payload_codec.go (revision 0) @@ -0,0 +1,116 @@ +/* + * Copyright (c) 2011 NeuStar, Inc. + * All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * NeuStar, the Neustar logo and related names and logos are registered + * trademarks, service marks or tradenames of NeuStar, Inc. All other + * product names, company names, marks, logos and symbols may be trademarks + * of their respective owners. + */ + +package kafka + +import ( + "bytes" + "compress/gzip" + // "log" +) + +const ( + NO_COMPRESSION_ID = 0 + GZIP_COMPRESSION_ID = 1 +) + +type PayloadCodec interface { + + // the 1 byte id of the codec + Id() byte + + // encoder interface for compression implementation + Encode(data []byte) []byte + + // decoder interface for decompression implementation + Decode(data []byte) []byte +} + +// Default Codecs + +var DefaultCodecs = []PayloadCodec{ + new(NoCompressionPayloadCodec), + new(GzipPayloadCodec), +} + +var DefaultCodecsMap = codecsMap(DefaultCodecs) + +func codecsMap(payloadCodecs []PayloadCodec) map[byte]PayloadCodec { + payloadCodecsMap := make(map[byte]PayloadCodec, len(payloadCodecs)) + for _, c := range payloadCodecs { + payloadCodecsMap[c.Id()] = c, true + } + return payloadCodecsMap +} + +// No compression codec, noop + +type NoCompressionPayloadCodec struct { + +} + +func (codec *NoCompressionPayloadCodec) Id() byte { + return NO_COMPRESSION_ID +} + +func (codec *NoCompressionPayloadCodec) Encode(data []byte) []byte { + return data +} + +func (codec *NoCompressionPayloadCodec) Decode(data []byte) []byte { + return data +} + +// Gzip Codec + +type GzipPayloadCodec struct { + +} + +func (codec *GzipPayloadCodec) Id() byte { + return GZIP_COMPRESSION_ID +} + +func (codec *GzipPayloadCodec) Encode(data []byte) []byte { + buf := bytes.NewBuffer([]byte{}) + zipper, _ := gzip.NewWriterLevel(buf, gzip.BestSpeed) + zipper.Write(data) + zipper.Close() + return buf.Bytes() +} + +func (codec *GzipPayloadCodec) Decode(data []byte) []byte { + buf := bytes.NewBuffer([]byte{}) + zipper, _ := gzip.NewReader(bytes.NewBuffer(data)) + unzipped := make([]byte, 100) + for { + n, err := zipper.Read(unzipped) + if n > 0 && err == nil { + buf.Write(unzipped[0:n]) + } else { + break + } + } + + zipper.Close() + return buf.Bytes() +} Index: src/timing.go =================================================================== --- src/timing.go (revision 1186859) +++ src/timing.go (working copy) @@ -20,7 +20,6 @@ * of their respective owners. */ - package kafka import ( Index: src/request.go =================================================================== --- src/request.go (revision 1186859) +++ src/request.go (working copy) @@ -28,19 +28,17 @@ "container/list" ) - type RequestType uint16 // Request Types const ( REQUEST_PRODUCE RequestType = 0 - REQUEST_FETCH = 1 - REQUEST_MULTIFETCH = 2 - REQUEST_MULTIPRODUCE = 3 - REQUEST_OFFSETS = 4 + REQUEST_FETCH = 1 + REQUEST_MULTIFETCH = 2 + REQUEST_MULTIPRODUCE = 3 + REQUEST_OFFSETS = 4 ) - // Request Header: func (b *Broker) EncodeRequestHeader(requestType RequestType) *bytes.Buffer { request := bytes.NewBuffer([]byte{}) @@ -70,7 +68,6 @@ return request.Bytes() } - // func (b *Broker) EncodeConsumeRequest(offset uint64, maxSize uint32) []byte { request := b.EncodeRequestHeader(REQUEST_FETCH) @@ -83,7 +80,6 @@ return request.Bytes() } - // func (b *Broker) EncodePublishRequest(messages *list.List) []byte { // 4 + 2 + 2 + topicLength + 4 + 4 @@ -103,6 +99,5 @@ binary.BigEndian.PutUint32(request.Bytes()[messageSetSizePos:], uint32(written)) // now add the size of the whole to the first uint32 encodeRequestSize(request) - return request.Bytes() } Index: README.md =================================================================== --- README.md (revision 1186859) +++ README.md (working copy) @@ -48,6 +48,17 @@ + +### Publishing Compressed Messages ### + +

+
+broker := kafka.NewBrokerPublisher("localhost:9092", "mytesttopic", 0)
+broker.Publish(kafka.NewCompressedMessage([]byte("tesing 1 2 3")))
+
+
+ + ### Consumer ### 

Index: Makefile
===================================================================
--- Makefile	(revision 1186859)
+++ Makefile	(working copy)
@@ -6,6 +6,7 @@
 	src/message.go\
 	src/converts.go\
 	src/consumer.go\
+	src/payload_codec.go\
 	src/publisher.go\
 	src/timing.go\
 	src/request.go\