initial commit

.devcontainer/devcontainer.json

@@ -0,0 +1,17 @@
+{
+	"name": "GOLANG Web Service",
+	"dockerComposeFile": "docker-compose.yml",
+	"workspaceFolder": "/workspaces",
+	"service": "go-web",
+	"remoteUser": "willes",
+
+	// Specify the services you want to launch, comment will launch all
+	"runServices": ["go-web", "zookeeper", "kafka1", "kafka2"],
+
+	"settings": {
+		"terminal.integrated.shell.linux": "/bin/bash"
+	},
+	"extensions": [
+		"dbaeumer.vscode-eslint"
+	],
+  }

.devcontainer/docker-compose.yml

@@ -0,0 +1,54 @@
+---
+version: '3'
+services:
+  go-web:
+    build:
+      context: .
+      dockerfile: go.dockerfile
+    command: ["tail -f /dev/null"]
+    environment:
+      - CONFIG_PATH=/workspaces/configs/server.json
+      - PORT=8000
+    volumes:
+      - ~/.ssh/id_rsa:/home/willes/.ssh/id_rsa
+      - ..:/workspaces
+      - ../../vscodeExtensions/extensions:/home/willes/.vscode-server/extensions
+      - ../../vscodeExtensions/insiders:/home/willes/.vscode-server-insiders/extensions
+    ports:
+      - 8100:8000
+
+  zookeeper:
+    image: wurstmeister/zookeeper
+    ports:
+      - "2181"
+    hostname: zookeeper
+
+  kafka1:
+    image: wurstmeister/kafka
+    command: [start-kafka.sh]
+    ports:
+      - 9092:9092
+    hostname: kafka1
+    environment:
+      KAFKA_ADVERTISED_HOST_NAME: kafka1 # docker-machine ip
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_ADVERTISED_PORT: 9092
+    volumes:
+      - /var/run/docker.sock:/var/run/docker.sock
+    depends_on:
+      - "zookeeper"
+
+  kafka2:
+    image: wurstmeister/kafka
+    command: [start-kafka.sh]
+    ports:
+      - 9093:9092
+    hostname: kafka2
+    environment:
+      KAFKA_ADVERTISED_HOST_NAME: kafka2 # docker-machine ip
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_ADVERTISED_PORT: 9092
+    volumes:
+      - /var/run/docker.sock:/var/run/docker.sock
+    depends_on:
+      - "zookeeper"

.devcontainer/go.dockerfile

@@ -0,0 +1,87 @@
+#-------------------------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See https://go.microsoft.com/fwlink/?linkid=2090316 for license information.
+#-------------------------------------------------------------------------------------------------------------
+
+FROM golang:1.14.2
+
+# Avoid warnings by switching to noninteractive
+ENV DEBIAN_FRONTEND=noninteractive
+
+# This Dockerfile adds a non-root user with sudo access. Use the "remoteUser"
+# property in devcontainer.json to use it. On Linux, the container user's GID/UIDs
+# will be updated to match your local UID/GID (when using the dockerFile property).
+# See https://aka.ms/vscode-remote/containers/non-root-user for details.
+ARG USERNAME=willes
+ARG USER_UID=1000
+ARG USER_GID=$USER_UID
+
+# Configure apt, install packages and tools
+RUN apt-get update \
+    && apt-get -y install --no-install-recommends apt-utils dialog 2>&1 \
+    #
+    # Verify git, process tools, lsb-release (common in install instructions for CLIs) installed
+    && apt-get -y install git openssh-client less iproute2 procps lsb-release \
+    #
+    # Build Go tools w/module support
+    && mkdir -p /tmp/gotools \
+    && cd /tmp/gotools \
+    && GOPATH=/tmp/gotools GO111MODULE=on go get -v golang.org/x/tools/gopls@latest 2>&1 \
+    && GOPATH=/tmp/gotools GO111MODULE=on go get -v \
+        honnef.co/go/tools/...@latest \
+        golang.org/x/tools/cmd/gorename@latest \
+        golang.org/x/tools/cmd/goimports@latest \
+        golang.org/x/tools/cmd/guru@latest \
+        golang.org/x/lint/golint@latest \
+        github.com/mdempsky/gocode@latest \
+        github.com/cweill/gotests/...@latest \
+        github.com/haya14busa/goplay/cmd/goplay@latest \
+        github.com/sqs/goreturns@latest \
+        github.com/josharian/impl@latest \
+        github.com/davidrjenni/reftools/cmd/fillstruct@latest \
+        github.com/uudashr/gopkgs/v2/cmd/gopkgs@latest  \
+        github.com/ramya-rao-a/go-outline@latest  \
+        github.com/acroca/go-symbols@latest  \
+        github.com/godoctor/godoctor@latest  \
+        github.com/rogpeppe/godef@latest  \
+        github.com/zmb3/gogetdoc@latest \
+        github.com/fatih/gomodifytags@latest  \
+        github.com/mgechev/revive@latest  \
+        github.com/go-delve/delve/cmd/dlv@latest 2>&1 \
+    #
+    # Build Go tools w/o module support
+    && GOPATH=/tmp/gotools go get -v github.com/alecthomas/gometalinter 2>&1 \
+    #
+    # Build gocode-gomod
+    && GOPATH=/tmp/gotools go get -x -d github.com/stamblerre/gocode 2>&1 \
+    && GOPATH=/tmp/gotools go build -o gocode-gomod github.com/stamblerre/gocode \
+    #
+    # Install Go tools
+    && mv /tmp/gotools/bin/* /usr/local/bin/ \
+    && mv gocode-gomod /usr/local/bin/ \
+    #
+    # Install golangci-lint
+    && curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b /usr/local/bin 2>&1 \
+    #
+    # Create a non-root user to use if preferred - see https://aka.ms/vscode-remote/containers/non-root-user.
+    && groupadd --gid $USER_GID $USERNAME \
+    && useradd -s /bin/bash --uid $USER_UID --gid $USER_GID -m $USERNAME \
+    # [Optional] Add sudo support
+    && apt-get install -y sudo \
+    && echo $USERNAME ALL=\(root\) NOPASSWD:ALL > /etc/sudoers.d/$USERNAME \
+    && chmod 0440 /etc/sudoers.d/$USERNAME \
+    #
+    # Clean up
+    && apt-get autoremove -y \
+    && apt-get clean -y \
+    && rm -rf /var/lib/apt/lists/* /tmp/gotools
+
+# Update this to "on" or "off" as appropriate
+ENV GO111MODULE=auto
+ENV DEBIAN_FRONTEND=dialog
+ENV GOPATH=/go
+
+RUN mkdir -p /home/$USERNAME/.vscode-server/extensions /home/$USERNAME/.vscode-server-insiders/extensions \
+    && chown -R $USERNAME /home/$USERNAME/.vscode-server /home/$USERNAME/.vscode-server-insiders
+
+ENTRYPOINT [ "/bin/sh", "-c" ]

.gitignore

@@ -13,3 +13,4 @@
 
 # Dependency directories (remove the comment below to include it)
 # vendor/
+.DS_Store

README.md

@@ -1 +1,33 @@
-# kafka-producer
+# kafka-producer
+
+
+# helm chart command to deploy cassandra
+
+# helm chart command to deploy kafka
+helm upgrade --install kafka --namespace kafka incubator/kafka
+
+# Commands
+Open the container in vscode remote container and run `go build`
+
+To launch the server, use `./kafka-producer <broker> <topic>`
+
+To publish, perform
+```
+curl --request POST 'http://localhost:8100/produce' \
+--header 'Content-Type: application/json' \
+--data-raw '{
+    "message": "data"
+}'
+```
+
+# Simple comparison
+On local laptop, running 50000 iterations
+
+No goroutine, 50 connections
+![No goroutine, 50 connections](./results/no_goroutine_50_concurrent.png)
+
+With goroutine, 50 connections
+![With goroutine, 50 connections](./results/goroutine_50_concurrent.png)
+
+With goroutine, 100 connections
+![With goroutine, 100 connections](./results/goroutine_100_concurrent.png)

eventqueue/kafka.go

@@ -0,0 +1,45 @@
+package eventqueue
+
+import (
+	"gopkg.in/confluentinc/confluent-kafka-go.v1/kafka"
+)
+
+// KafkaConfig the config to kafka
+type KafkaConfig struct {
+	Broker string
+}
+
+type kafkaProducer struct {
+	producer     *kafka.Producer
+	deliveryChan chan kafka.Event
+}
+
+func (k *kafkaProducer) Produce(topic string, message string) error {
+	k.producer.Produce(&kafka.Message{
+		TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
+		Value:          []byte(message),
+	}, k.deliveryChan)
+
+	e := <-k.deliveryChan
+	m := e.(*kafka.Message)
+
+	if m.TopicPartition.Error != nil {
+		return m.TopicPartition.Error
+	}
+	return nil
+}
+
+func (k *kafkaProducer) Close() {
+	close(k.deliveryChan)
+}
+
+// NewProducer constructor function to kafka producer
+func NewProducer(config *KafkaConfig) (Producer, error) {
+	p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": config.Broker})
+	if err != nil {
+		return nil, err
+	}
+
+	k := kafkaProducer{p, make(chan kafka.Event)}
+	return &k, nil
+}

eventqueue/producer.go

@@ -0,0 +1,7 @@
+package eventqueue
+
+// Producer the producer interface
+type Producer interface {
+	Produce(topic string, message string) error
+	Close()
+}

go.mod

@@ -0,0 +1,8 @@
+module github.com/willeslau/kafka-producer
+
+go 1.14
+
+require (
+	github.com/confluentinc/confluent-kafka-go v1.4.2 // indirect
+	gopkg.in/confluentinc/confluent-kafka-go.v1 v1.4.2
+)

go.sum

@@ -0,0 +1,4 @@
+github.com/confluentinc/confluent-kafka-go v1.4.2 h1:13EK9RTujF7lVkvHQ5Hbu6bM+Yfrq8L0MkJNnjHSd4Q=
+github.com/confluentinc/confluent-kafka-go v1.4.2/go.mod h1:u2zNLny2xq+5rWeTQjFHbDzzNuba4P1vo31r9r4uAdg=
+gopkg.in/confluentinc/confluent-kafka-go.v1 v1.4.2 h1:JabkIV98VYFqYKHHzXtgGMFuRgFBNTNzBytbGByzrJI=
+gopkg.in/confluentinc/confluent-kafka-go.v1 v1.4.2/go.mod h1:ZdI3yfYmdNSLQPNCpO1y00EHyWaHG5EnQEyL/ntAegY=

main.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+
+	"github.com/willeslau/kafka-producer/eventqueue"
+)
+
+func main() {
+	broker := os.Args[1]
+	topic := os.Args[2]
+
+	config := eventqueue.KafkaConfig{broker}
+	producer, err := eventqueue.NewProducer(&config)
+	if err != nil {
+		panic(err)
+	}
+	defer producer.Close()
+
+	http.HandleFunc("/produce", getHandler(producer, topic))
+	log.Fatal(http.ListenAndServe(":8000", nil))
+}
+
+// Payload the payload
+type Payload struct {
+	Message string
+}
+
+func getHandler(p eventqueue.Producer, topic string) func(w http.ResponseWriter, r *http.Request) {
+	fn := func(w http.ResponseWriter, r *http.Request) {
+		payload := Payload{}
+		err := json.NewDecoder(r.Body).Decode(&payload)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			fmt.Fprintf(w, "Error!")
+			return
+		}
+
+		p.Produce(topic, payload.Message)
+		fmt.Fprintf(w, "Hi there, I love %s!", r.URL.Path[1:])
+	}
+	return fn
+}