1 // Writing a basic HTTP server is easy using the
16 "github.com/confluentinc/confluent-kafka-go/kafka"
17 "github.com/enriquebris/goconcurrentqueue"
18 "github.com/gorilla/mux"
21 // Note: consumer 'group' and consumer 'user' both uses hardcoded values specific to this interface
22 // globalCounters var holding the "global counters"
23 // recieved number of received messages from all topics (int)
24 // sent number of sent messages to all topics (int)
25 // topics var holding all topic related info
26 // <topic-name> name of a topic (present after topic is created)
27 // content-type data type of the topic (string)
29 // recieved number of received messages from the topic (int)
30 // sent number of sent messages to the topic (int)
32 // send messages waiting to be sent (set when sending is started) (fifo)
33 // received received messages waiting to be fetched (set when reception is started) (fifo)
39 func (c *counter) step() {
40 atomic.AddUint64(&c.c, 1)
43 func (c counter) get() uint64 {
44 return atomic.LoadUint64(&c.c)
47 type counters struct {
52 func newCounters() counters {
59 type messages struct {
60 send *goconcurrentqueue.FIFO
61 received *goconcurrentqueue.FIFO
64 func (m *messages) startSend() bool {
66 m.send = goconcurrentqueue.NewFIFO()
72 func (m *messages) stopSend() {
76 func (m *messages) addToSend(msg string) error {
78 return fmt.Errorf("sending not started")
82 return m.send.Enqueue(msg)
85 func (m *messages) getToSend() (interface{}, error) {
87 return "", fmt.Errorf("sending not started")
91 return m.send.Dequeue()
94 func (m *messages) startReceive() bool {
95 if m.received == nil {
96 m.received = goconcurrentqueue.NewFIFO()
102 func (m *messages) stopReceive() {
112 func newTopic(ct string) *topic {
115 counters: counters{},
116 messages: messages{},
120 var globalCounters counters
121 var topics map[string]*topic = make(map[string]*topic)
123 var bootstrapserver = ""
126 bootstrapserver = os.Getenv("KAFKA_BOOTSTRAP_SERVER")
127 if len(bootstrapserver) == 0 {
128 fmt.Println("Fatal error: env var KAFKA_BOOTSTRAP_SERVER not set")
129 fmt.Println("Exiting... ")
132 fmt.Println("Using KAFKA_BOOTSTRAP_SERVER=" + bootstrapserver)
135 //Helper function to get a created topic, if it exists
136 func getTopicFromRequest(w http.ResponseWriter, req *http.Request) (*topic, string, bool) {
137 topicId := mux.Vars(req)["topic"]
138 t, exist := topics[topicId]
140 w.WriteHeader(http.StatusNotFound)
141 fmt.Fprintf(w, "Topic %v does not exist", topicId)
142 return nil, "", false
144 return t, topicId, true
149 func healthCheck(w http.ResponseWriter, req *http.Request) {
153 // Deep reset of this interface stub - no removal of msgs or topics in kafka
155 func allreset(w http.ResponseWriter, req *http.Request) {
156 for _, v := range topics {
157 v.messages.stopSend()
158 v.messages.stopReceive()
160 time.Sleep(5 * time.Second) //Allow producers/consumers to shut down
161 globalCounters = newCounters()
162 topics = make(map[string]*topic)
166 // Get topics, return json array of strings of topics created by this interface stub
167 // Returns json array of strings, array is empty if no topics exist
170 func getTopics(w http.ResponseWriter, req *http.Request) {
171 topicKeys := make([]string, 0, len(topics))
172 fmt.Printf("len topics: %v\n", len(topics))
173 for k := range topics {
174 topicKeys = append(topicKeys, k)
176 var j, err = json.Marshal(topicKeys)
178 w.WriteHeader(http.StatusInternalServerError)
179 fmt.Fprintf(w, "Cannot convert list of topics to json, error details: %v", err)
182 w.Header().Set("Content-Type", "application/json")
183 w.WriteHeader(http.StatusOK)
188 func writeOkRepsonse(w http.ResponseWriter, httpStatus int, msg string) {
189 w.WriteHeader(httpStatus)
190 w.Header().Set("Content-Type", "text/plain")
194 // Get a counter value
195 // GET /topics/counters/{counter}
196 func getCounter(w http.ResponseWriter, req *http.Request) {
197 ctr := mux.Vars(req)["counter"]
199 if ctr == "received" {
200 ctrvalue = int(globalCounters.received.get())
201 } else if ctr == "sent" {
202 ctrvalue = int(globalCounters.sent.get())
206 w.WriteHeader(http.StatusBadRequest)
207 fmt.Fprintf(w, "Counter %v does not exist", ctr)
210 writeOkRepsonse(w, http.StatusOK, strconv.Itoa(ctrvalue))
216 // PUT on /topics/<topic>?type=<type> type shall be 'application/json' or 'text/plain'
217 func createTopic(w http.ResponseWriter, req *http.Request) {
218 topicId := mux.Vars(req)["topic"]
219 topicType := mux.Vars(req)["type"]
221 fmt.Printf("Creating topic: %v, content type: %v\n", topicId, topicType)
223 if len(topicType) == 0 {
224 w.WriteHeader(http.StatusBadRequest)
225 fmt.Fprintf(w, "Type not specified")
229 tid, exist := topics[topicId]
231 if tid.contentType != topicType {
232 w.WriteHeader(http.StatusBadRequest)
233 fmt.Fprintf(w, "Topic type exist but type is different, queue content type: %v, requested content type: %v", tid.contentType, topicType)
236 writeOkRepsonse(w, http.StatusOK, "Topic exist")
240 t := newTopic(topicType)
242 a, err := kafka.NewAdminClient(&kafka.ConfigMap{"bootstrap.servers": bootstrapserver})
243 defer func() { a.Close() }()
245 w.WriteHeader(http.StatusInternalServerError)
246 fmt.Fprintf(w, "Cannot create client to bootstrap server: "+bootstrapserver+", error details: %v", err)
250 ctx, cancel := context.WithCancel(context.Background())
253 maxDur := 10 * time.Second
255 _, err = a.CreateTopics(
257 []kafka.TopicSpecification{{
260 ReplicationFactor: 1}},
261 kafka.SetAdminOperationTimeout(maxDur))
264 w.WriteHeader(http.StatusInternalServerError)
265 fmt.Fprintf(w, "Failed to create topic: %v, error details: %v", topicId, err)
269 w.WriteHeader(http.StatusCreated)
270 fmt.Fprintf(w, "Topic created")
274 // GET on /topic/<topic>
275 func getTopic(w http.ResponseWriter, req *http.Request) {
276 t, _, exist := getTopicFromRequest(w, req)
280 w.WriteHeader(http.StatusOK)
281 fmt.Fprintf(w, t.contentType)
284 // Get a topics counter value
285 // GET /topics/{topic}/counters/{counter}
286 func getTopicCounter(w http.ResponseWriter, req *http.Request) {
287 t, _, exist := getTopicFromRequest(w, req)
291 ctr := mux.Vars(req)["counter"]
294 if ctr == "received" {
295 ctrvalue = int(t.counters.received.get())
296 } else if ctr == "sent" {
297 ctrvalue = int(t.counters.sent.get())
301 w.WriteHeader(http.StatusBadRequest)
302 fmt.Fprintf(w, "Counter %v does not exist", ctr)
305 w.WriteHeader(http.StatusOK)
306 fmt.Fprintf(w, strconv.Itoa(ctrvalue))
310 func startToSend(w http.ResponseWriter, req *http.Request) {
311 t, topicId, exist := getTopicFromRequest(w, req)
316 if !t.messages.startSend() {
317 w.WriteHeader(http.StatusOK)
318 fmt.Fprintf(w, "Already started sending")
322 p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": bootstrapserver})
324 fmt.Printf("Cannot create producer for topic: %v, error details: %v\n", topicId, err)
327 defer func() { p.Close() }()
335 err = p.Produce(&kafka.Message{
336 TopicPartition: kafka.TopicPartition{Topic: &topicId, Partition: kafka.PartitionAny},
337 Value: []byte(fmt.Sprintf("%v", m)),
341 t.counters.sent.step()
342 globalCounters.sent.step()
343 msg := fmt.Sprintf("%v", m)
345 fmt.Printf("Message sent on topic: %v, len: %v, msg: %v", topicId, len(msg), msg)
347 fmt.Printf("Message sent on topic: %v, len: %v, is larger than 500...not printed", topicId, len(msg))
350 fmt.Printf("Failed to send message on topic: %v. Discarded. Error details: %v", topicId, err)
354 time.Sleep(10 * time.Millisecond)
359 w.WriteHeader(http.StatusOK)
360 fmt.Fprintf(w, "Sending started")
363 func startToReceive(w http.ResponseWriter, req *http.Request) {
364 t, topicId, exist := getTopicFromRequest(w, req)
369 if !t.messages.startReceive() {
370 w.WriteHeader(http.StatusOK)
371 fmt.Fprintf(w, "Already started receiving")
377 defer func() { t.messages.stopReceive() }()
379 groudId := "kafkaprocon"
381 c, err := kafka.NewConsumer(&kafka.ConfigMap{
382 "bootstrap.servers": bootstrapserver,
384 "auto.offset.reset": "earliest",
385 "enable.auto.commit": true,
386 "auto.commit.interval.ms": 5000,
389 fmt.Printf("Cannot create consumer for topic: %v, error details: %v\n", topicId, err)
390 t.messages.stopReceive()
394 defer func() { c.Close() }()
396 que := t.messages.received
398 fmt.Printf("Cannot start receiving on topic: %v, queue does not exist\n", topicId)
401 fmt.Printf("Start subscribing on topic: %v\n", topicId)
402 err = c.SubscribeTopics([]string{topicId}, nil)
404 fmt.Printf("Cannot start subscribing on topic: %v, error details: %v\n", topicId, err)
407 maxDur := 1 * time.Second
409 msg, err := c.ReadMessage(maxDur)
411 if len(msg.Value) < 500 {
412 fmt.Printf("Message received on topic: %v, partion: %v, len: %v, msg: %v", topicId, msg.TopicPartition, len(msg.Value), string(msg.Value))
414 fmt.Printf("Message received on topic: %v, partion: %v, len: %v is larger than 500...not printed", topicId, msg.TopicPartition, len(msg.Value))
416 err = t.messages.received.Enqueue(string(msg.Value))
418 w.WriteHeader(http.StatusInternalServerError)
419 fmt.Fprintf(w, "Received message topic: %v, cannot be put in queue, %v", topicId, err)
422 t.counters.received.step()
423 globalCounters.received.step()
425 fmt.Printf("Nothing to consume on topic: %v, reason: %v\n", topicId, err)
431 w.WriteHeader(http.StatusOK)
432 fmt.Fprintf(w, "Receiving started")
435 // Post a message to a topic
436 // POST /send content type is specified in content type
437 func sendToTopic(w http.ResponseWriter, req *http.Request) {
438 t, topicId, exist := getTopicFromRequest(w, req)
444 w.WriteHeader(http.StatusBadRequest)
445 fmt.Fprintf(w, "Sending not initiated on topic: %v", topicId)
448 ct := req.Header.Get("Content-Type")
449 if ct != t.contentType {
450 w.WriteHeader(http.StatusBadRequest)
451 fmt.Fprintf(w, "Message to send content type: %v on topic: %v does not match queue content type: %v", ct, topicId, t.contentType)
455 if ct == "application/json" {
456 // decoder := json.NewDecoder(req.Body)
458 // err := decoder.Decode(&j)
460 // w.WriteHeader(http.StatusBadRequest)
461 // w.Header().Set("Content-Type", "text/plain")
462 // fmt.Fprintf(w, "Json payload cannot be decoded, error details %v\n", err)
465 //m = mux.Vars(req)[""]
466 if err := req.ParseForm(); err != nil {
467 w.WriteHeader(http.StatusBadRequest)
468 fmt.Fprintf(w, "Json payload cannot be decoded on topic: %v, error details %v", topicId, err)
471 b, err := ioutil.ReadAll(req.Body)
474 fmt.Printf("Json payload to send on topic: %v, msg: %v", topicId, string(b))
476 fmt.Printf("Json payload to send on topic: %v larger than 500 bytes, not printed...", topicId)
479 err = q.Enqueue(string(b))
481 w.WriteHeader(http.StatusInternalServerError)
482 fmt.Fprintf(w, "Json message to send cannot be put in queue")
485 } else if ct == "text/plain" {
486 if err := req.ParseForm(); err != nil {
487 w.WriteHeader(http.StatusBadRequest)
488 fmt.Fprintf(w, "Text payload to send on topic: %v cannot be decoded, error details %v\n", topicId, err)
491 b, err := ioutil.ReadAll(req.Body)
494 fmt.Printf("Text payload to send on topic: %v, msg: %v", topicId, string(b))
496 fmt.Printf("Text payload to send on topic: %v larger than 500 bytes, not printed...", topicId)
499 err = q.Enqueue(string(b))
501 w.WriteHeader(http.StatusInternalServerError)
502 fmt.Fprintf(w, "Text message to send cannot be put in queue")
506 w.WriteHeader(http.StatusBadRequest)
507 fmt.Fprintf(w, "Message to send, unknown content type %v", ct)
511 w.WriteHeader(http.StatusOK)
512 w.Header().Set("Content-Type", "text/plain")
513 fmt.Fprintf(w, "Message to send put in queue")
516 // Get zero or one message from a topic
518 func receiveFromTopic(w http.ResponseWriter, req *http.Request) {
519 t, topicId, exist := getTopicFromRequest(w, req)
523 if t.messages.received == nil {
524 w.WriteHeader(http.StatusBadRequest)
525 fmt.Fprintf(w, "Receiving not initiated on topic %v", topicId)
529 m, err := t.messages.received.Dequeue()
531 w.WriteHeader(http.StatusNoContent)
535 w.Header().Set("Content-Type", t.contentType)
536 w.WriteHeader(http.StatusOK)
537 fmt.Fprintf(w, "%v", m)
540 // Remove the send queue to stop sending
541 func stopToSend(w http.ResponseWriter, req *http.Request) {
542 fmt.Printf("Stop sending\n")
543 t, _, exist := getTopicFromRequest(w, req)
547 t.messages.stopSend()
548 w.WriteHeader(http.StatusNoContent)
551 // Remove the receive queue to stop receiving
552 func stopToReceive(w http.ResponseWriter, req *http.Request) {
553 fmt.Printf("Stop receiving\n")
554 t, _, exist := getTopicFromRequest(w, req)
558 t.messages.stopReceive()
559 w.WriteHeader(http.StatusNoContent)
562 func HelloServer(w http.ResponseWriter, r *http.Request) {
563 fmt.Fprintf(w, "Hello, %s!", r.URL.Path[1:])
572 r.HandleFunc("/", healthCheck).Methods("GET")
573 r.HandleFunc("/reset", allreset).Methods("POST")
574 r.HandleFunc("/counters/{counter}", getCounter).Methods("GET")
575 r.HandleFunc("/topics", getTopics).Methods("GET")
576 r.HandleFunc("/topics/{topic}/counters/{counter}", getTopicCounter).Methods("GET")
577 r.HandleFunc("/topics/{topic}", createTopic).Methods("PUT").Queries("type", "{type}")
578 r.HandleFunc("/topics/{topic}", getTopic).Methods("GET")
579 r.HandleFunc("/topics/{topic}/startsend", startToSend).Methods("POST")
580 r.HandleFunc("/topics/{topic}/startreceive", startToReceive).Methods("POST")
581 r.HandleFunc("/topics/{topic}/stopsend", stopToSend).Methods("POST")
582 r.HandleFunc("/topics/{topic}/stopreceive", stopToReceive).Methods("POST")
583 r.HandleFunc("/topics/{topic}/msg", sendToTopic).Methods("POST")
584 r.HandleFunc("/topics/{topic}/msg", receiveFromTopic).Methods("GET")
590 WriteTimeout: 15 * time.Second,
591 ReadTimeout: 15 * time.Second,
593 fmt.Println("Running on port: " + port)
594 fmt.Printf(srv.ListenAndServe().Error())