From ecfba6e2090750238ad039064bbe4b56ee886efb Mon Sep 17 00:00:00 2001
From: fengzhiqiang <fengzhq@it2000.com.cn>
Date: Fri, 11 Jan 2019 14:26:25 +0800
Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=8F=91=E9=80=81=E6=B6=88?=
 =?UTF-8?q?=E6=81=AF=E7=9A=84=E5=8A=9F=E8=83=BD?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 serv.go | 38 ++++++++++++++++++++++++++++++++++++++
 1 file changed, 38 insertions(+)

diff --git a/serv.go b/serv.go
index d72ab9f..2b549de 100644
--- a/serv.go
+++ b/serv.go
@@ -5,6 +5,7 @@ import (
 	"context"
 	"flag"
 	"io"
+	"io/ioutil"
 	"net/http"
 	"os"
 	"os/signal"
@@ -337,6 +338,41 @@ func destroy() {
 	log.Warn(" terminated ")
 }
 
+func produceHandler(w http.ResponseWriter, r *http.Request) {
+	vars := mux.Vars(r)
+	topic := vars["topic"]
+	key := vars["key"]
+	msg, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		log.Error(err)
+		http.Error(w, err.Error(), 500)
+		return
+	}
+	writer := kafka.NewWriter(kafka.WriterConfig{
+		Brokers:  strings.Split(*bootstrapServers, ","),
+		Topic:    topic,
+		Balancer: &kafka.LeastBytes{},
+	})
+	log.WithFields(log.Fields{
+		"topic": topic,
+		"key":   key,
+		"msg":   string(msg),
+	}).Info("sending message")
+	senderr := writer.WriteMessages(context.Background(),
+		kafka.Message{
+			Key:   []byte(key),
+			Value: msg,
+		})
+	if senderr != nil {
+		log.Error(err)
+		http.Error(w, err.Error(), 500)
+		return
+	}
+	log.Info("send ok")
+	writer.Close()
+	w.WriteHeader(200)
+}
+
 func main() {
 	flag.Parse()
 	if *logfile != "" {
@@ -360,6 +396,8 @@ func main() {
 
 	router.Path("/ws/{topic}/{partition}").Queries("offset", "{offset}").HandlerFunc(websocketHandler).Name("web-socket")
 
+	router.Path("/produce/{topic}/{key}").HandlerFunc(produceHandler).Methods("POST")
+
 	// Run our server in a goroutine so that it doesn't block.
 	go func() {
 		log.WithFields(log.Fields{"address": *addr}).Info("starting server")