Initial checkpoint

client.go

@@ -0,0 +1,55 @@
+package kafka
+
+import (
+	"encoding/binary"
+	"net"
+)
+
+type Client struct {
+	addr string
+	conn net.Conn
+}
+
+func NewClient(addr string) (client *Client, err error) {
+	conn, err := net.Dial("tcp", addr)
+	if err != nil {
+		return nil, err
+	}
+	client = &Client{addr, conn}
+	return client, err
+}
+
+func (client *Client) write(buf []byte) (err error) {
+	var size [4]byte
+	binary.BigEndian.PutUint32(size[:], uint32(len(buf)))
+	_, err = client.conn.Write(size[:])
+	if err != nil {
+		return err
+	}
+	_, err = client.conn.Write(buf)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (client *Client) read() (buf []byte, err error) {
+	var size [4]byte
+	n, err := client.conn.Read(size[:])
+	if err != nil {
+		return nil, err
+	}
+	if n != 4 {
+		return nil, nil
+	}
+	s := binary.BigEndian.Uint32(size[:])
+	buf = make([]byte, s)
+	n, err = client.conn.Read(buf)
+	if err != nil {
+		return nil, err
+	}
+	if uint32(n) != s {
+		return nil, nil
+	}
+	return buf, nil
+}

consumer.go

@@ -0,0 +1,5 @@
+package kafka
+
+type Consumer struct {
+	client Client
+}

producer.go

@@ -0,0 +1,5 @@
+package kafka
+
+type Producer struct {
+	client Client
+}

request.go

@@ -0,0 +1,27 @@
+package kafka
+
+type RequestType uint16
+
+const (
+	REQUEST_PRODUCE        RequestType = 0
+	REQUEST_FETCH                      = 1
+	REQUEST_OFFSET                     = 2
+	REQUEST_METADATA                   = 3
+	REQUEST_LEADER_AND_ISR             = 4
+	REQUEST_STOP_REPLICA               = 5
+	REQUEST_OFFSET_COMMIT              = 6
+	REQUEST_OFFSET_FETCH               = 7
+)
+
+type ApiVersion uint16
+
+var API_VERSIONS = map[RequestType]ApiVersion{
+	REQUEST_PRODUCE:        0,
+	REQUEST_FETCH:          0,
+	REQUEST_OFFSET:         0,
+	REQUEST_METADATA:       0,
+	REQUEST_LEADER_AND_ISR: 0,
+	REQUEST_STOP_REPLICA:   0,
+	REQUEST_OFFSET_COMMIT:  0,
+	REQUEST_OFFSET_FETCH:   0,
+}