merged jdamick/kafka.go into the kafka tree to sit with the other cli…

…ents

clients/go/.gitignore

@@ -0,0 +1,8 @@
+_go_.6
+_obj
+tools/*/_obj
+tools/*/_go_.6
+tools/consumer/consumer
+tools/publisher/publisher
+tools/consumer/test.txt
+tools/offsets/offsets

clients/go/LICENSE

@@ -0,0 +1,24 @@
+Copyright (c) 2000-2011 NeuStar, Inc.
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+    * Redistributions of source code must retain the above copyright
+      notice, this list of conditions and the following disclaimer.
+    * Redistributions in binary form must reproduce the above copyright
+      notice, this list of conditions and the following disclaimer in the
+      documentation and/or other materials provided with the distribution.
+    * Neither the name of the NeuStar nor the
+      names of its contributors may be used to endorse or promote products
+      derived from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL NEUSTAR BE LIABLE FOR ANY
+DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

clients/go/Makefile

@@ -0,0 +1,25 @@
+include $(GOROOT)/src/Make.inc
+
+TARG=kafka
+GOFILES=\
+	src/kafka.go\
+	src/message.go\
+	src/converts.go\
+	src/consumer.go\
+	src/publisher.go\
+	src/timing.go\
+	src/request.go\
+
+include $(GOROOT)/src/Make.pkg
+
+tools: force
+	make -C tools/consumer clean all
+	make -C tools/publisher clean all
+	make -C tools/offsets clean all
+
+format:
+	gofmt -w -tabwidth=2 -tabindent=false src/*.go tools/consumer/*.go  tools/publisher/*.go kafka_test.go
+
+full: format clean install tools
+
+.PHONY: force 

clients/go/README.md

@@ -0,0 +1,74 @@
+# Kafka.go - Publisher & Consumer for Kafka in Go #
+
+Kafka is a distributed publish-subscribe messaging system: (http://sna-projects.com/kafka/)
+
+Go language: (http://golang.org/) <br/>
+
+## Get up and running ##
+
+Install kafka.go package: <br/>
+<code>make install</code>
+<br/>
+Make the tools (publisher & consumer) <br/>
+<code>make tools</code>
+<br/>
+Start zookeeper, Kafka server <br/>
+For more info on Kafka, see: http://sna-projects.com/kafka/quickstart.php
+
+
+
+## Tools ##
+
+Start a consumer:
+<pre><code>
+   ./tools/consumer/consumer -topic test -consumeforever
+  Consuming Messages :
+  From: localhost:9092, topic: test, partition: 0
+   ---------------------- 
+</code></pre>
+
+Now the consumer will just poll until a message is received.
+
+Publish a message:
+<pre><code>
+  ./tools/publisher/publisher -topic test -message "Hello World"
+</code></pre>
+
+The consumer should output message.
+
+## API Usage ##
+
+### Publishing ###
+
+
+<pre><code>
+
+broker := kafka.NewBrokerPublisher("localhost:9092", "mytesttopic", 0)
+broker.Publish(kafka.NewMessage([]byte("tesing 1 2 3")))
+
+</code></pre>
+
+### Consumer ###
+
+<pre><code>
+broker := kafka.NewBrokerConsumer("localhost:9092", "mytesttopic", 0, 0, 1048576)
+broker.Consume(func(msg *kafka.Message) { msg.Print() })
+
+</code></pre>
+
+Or the consumer can use a channel based approach:
+
+<pre><code>
+broker := kafka.NewBrokerConsumer("localhost:9092", "mytesttopic", 0, 0, 1048576)
+go broker.ConsumeOnChannel(msgChan, 10, quitChan)
+
+</code></pre>
+
+### Consuming Offsets ###
+
+<pre><code>
+broker := kafka.NewBrokerOffsetConsumer("localhost:9092", "mytesttopic", 0)
+offsets, err := broker.GetOffsets(-1, 1)
+</code></pre>
+
+

clients/go/kafka_test.go

@@ -0,0 +1,113 @@
+/*
+ * Copyright 2000-2011 NeuStar, Inc. All rights reserved.
+ * 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 (
+  "testing"
+  //"fmt"
+  "bytes"
+  "container/list"
+)
+
+func TestMessageCreation(t *testing.T) {
+  payload := []byte("testing")
+  msg := NewMessage(payload)
+  if msg.magic != 0 {
+    t.Errorf("magic incorrect")
+    t.Fail()
+  }
+
+  // generated by kafka-rb: e8 f3 5a 06
+  expected := []byte{0xe8, 0xf3, 0x5a, 0x06}
+  if !bytes.Equal(expected, msg.checksum[:]) {
+    t.Fail()
+  }
+}
+
+
+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}
+  if !bytes.Equal(expected, msg.Encode()) {
+    t.Fail()
+  }
+
+  // verify round trip
+  msgDecoded := Decode(msg.Encode())
+  if !bytes.Equal(msgDecoded.payload, payload) {
+    t.Fail()
+  }
+  if !bytes.Equal(msgDecoded.payload, payload) {
+    t.Fail()
+  }
+  chksum := []byte{0xE8, 0xF3, 0x5A, 0x06}
+  if !bytes.Equal(msgDecoded.checksum[:], chksum) {
+    t.Fail()
+  }
+  if msgDecoded.magic != 0 {
+    t.Fail()
+  }
+}
+
+func TestRequestHeaderEncoding(t *testing.T) {
+  broker := newBroker("localhost:9092", "test", 0)
+  request := broker.EncodeRequestHeader(REQUEST_PRODUCE)
+
+  // generated by kafka-rb:
+  expected := []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x04, 0x74, 0x65, 0x73, 0x74,
+    0x00, 0x00, 0x00, 0x00}
+
+  if !bytes.Equal(expected, request.Bytes()) {
+    t.Errorf("expected length: %d but got: %d", len(expected), len(request.Bytes()))
+    t.Errorf("expected: %X\n but got: %X", expected, request)
+    t.Fail()
+  }
+}
+
+
+func TestPublishRequestEncoding(t *testing.T) {
+  payload := []byte("testing")
+  msg := NewMessage(payload)
+
+  messages := list.New()
+  messages.PushBack(msg)
+  pubBroker := NewBrokerPublisher("localhost:9092", "test", 0)
+  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}
+
+  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.Fail()
+  }
+}
+
+func TestConsumeRequestEncoding(t *testing.T) {
+
+  pubBroker := NewBrokerPublisher("localhost:9092", "test", 0)
+  request := pubBroker.broker.EncodeConsumeRequest(0, 1048576)
+
+  // generated by kafka-rb, encode_request_size + encode_request
+  expected := []byte{0x00, 0x00, 0x00, 0x18, 0x00, 0x01, 0x00, 0x04, 0x74,
+    0x65, 0x73, 0x74, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+    0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x10, 0x00, 0x00}
+
+  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.Fail()
+  }
+}