Add usecases to project pom
[nonrtric.git] / test / kafka-procon / main.go
1 // Writing a basic HTTP server is easy using the
2 // `net/http` package.
3 package main
4
5 import (
6         "context"
7         "encoding/json"
8         "fmt"
9         "io/ioutil"
10         "net/http"
11         "os"
12         "strconv"
13         "sync/atomic"
14         "time"
15
16         "github.com/confluentinc/confluent-kafka-go/kafka"
17         "github.com/enriquebris/goconcurrentqueue"
18         "github.com/gorilla/mux"
19 )
20
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)
28 //        counters
29 //          recieved      number of received messages from the topic                              (int)
30 //          sent          number of sent messages to the topic                                    (int)
31 //        messages
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)
34
35 type counter struct {
36         c uint64
37 }
38
39 func (c *counter) step() {
40         atomic.AddUint64(&c.c, 1)
41 }
42
43 func (c counter) get() uint64 {
44         return atomic.LoadUint64(&c.c)
45 }
46
47 type counters struct {
48         received counter
49         sent     counter
50 }
51
52 func newCounters() counters {
53         return counters{
54                 received: counter{},
55                 sent:     counter{},
56         }
57 }
58
59 type messages struct {
60         send     *goconcurrentqueue.FIFO
61         received *goconcurrentqueue.FIFO
62 }
63
64 func (m *messages) startSend() bool {
65         if m.send == nil {
66                 m.send = goconcurrentqueue.NewFIFO()
67                 return true
68         }
69         return false
70 }
71
72 func (m *messages) stopSend() {
73         m.send = nil
74 }
75
76 func (m *messages) addToSend(msg string) error {
77         if m.send == nil {
78                 return fmt.Errorf("sending not started")
79         }
80         m.send.Lock()
81         defer m.send.Unlock()
82         return m.send.Enqueue(msg)
83 }
84
85 func (m *messages) getToSend() (interface{}, error) {
86         if m.send == nil {
87                 return "", fmt.Errorf("sending not started")
88         }
89         m.send.Lock()
90         defer m.send.Unlock()
91         return m.send.Dequeue()
92 }
93
94 func (m *messages) startReceive() bool {
95         if m.received == nil {
96                 m.received = goconcurrentqueue.NewFIFO()
97                 return true
98         }
99         return false
100 }
101
102 func (m *messages) stopReceive() {
103         m.send = nil
104 }
105
106 type topic struct {
107         contentType string
108         counters    counters
109         messages    messages
110 }
111
112 func newTopic(ct string) *topic {
113         return &topic{
114                 contentType: ct,
115                 counters:    counters{},
116                 messages:    messages{},
117         }
118 }
119
120 var globalCounters counters
121 var topics map[string]*topic = make(map[string]*topic)
122
123 var bootstrapserver = ""
124
125 func initApp() {
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... ")
130                 os.Exit(1)
131         }
132         fmt.Println("Using KAFKA_BOOTSTRAP_SERVER=" + bootstrapserver)
133 }
134
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]
139         if exist == false {
140                 w.WriteHeader(http.StatusNotFound)
141                 fmt.Fprintf(w, "Topic %v does not exist", topicId)
142                 return nil, "", false
143         }
144         return t, topicId, true
145 }
146
147 // Alive check
148 // GET on /
149 func healthCheck(w http.ResponseWriter, req *http.Request) {
150         fmt.Fprintf(w, "OK")
151 }
152
153 // Deep reset of this interface stub - no removal of msgs or topics in kafka
154 // POST on /reset
155 func allreset(w http.ResponseWriter, req *http.Request) {
156         for _, v := range topics {
157                 v.messages.stopSend()
158                 v.messages.stopReceive()
159         }
160         time.Sleep(5 * time.Second) //Allow producers/consumers to shut down
161         globalCounters = newCounters()
162         topics = make(map[string]*topic)
163         fmt.Fprintf(w, "OK")
164 }
165
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
168 // GET on /topics
169
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)
175         }
176         var j, err = json.Marshal(topicKeys)
177         if err != nil {
178                 w.WriteHeader(http.StatusInternalServerError)
179                 fmt.Fprintf(w, "Cannot convert list of topics to json, error details: %v", err)
180                 return
181         } else {
182                 w.Header().Set("Content-Type", "application/json")
183                 w.WriteHeader(http.StatusOK)
184                 w.Write(j)
185         }
186 }
187
188 func writeOkRepsonse(w http.ResponseWriter, httpStatus int, msg string) {
189         w.WriteHeader(httpStatus)
190         w.Header().Set("Content-Type", "text/plain")
191         fmt.Fprintf(w, msg)
192 }
193
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"]
198         var ctrvalue = -1
199         if ctr == "received" {
200                 ctrvalue = int(globalCounters.received.get())
201         } else if ctr == "sent" {
202                 ctrvalue = int(globalCounters.sent.get())
203         }
204
205         if ctrvalue == -1 {
206                 w.WriteHeader(http.StatusBadRequest)
207                 fmt.Fprintf(w, "Counter %v does not exist", ctr)
208                 return
209         }
210         writeOkRepsonse(w, http.StatusOK, strconv.Itoa(ctrvalue))
211         return
212
213 }
214
215 // Create a topic
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"]
220
221         fmt.Printf("Creating topic: %v, content type: %v\n", topicId, topicType)
222
223         if len(topicType) == 0 {
224                 w.WriteHeader(http.StatusBadRequest)
225                 fmt.Fprintf(w, "Type not specified")
226                 return
227         }
228
229         tid, exist := topics[topicId]
230         if exist == true {
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)
234                         return
235                 }
236                 writeOkRepsonse(w, http.StatusOK, "Topic exist")
237                 return
238         }
239
240         t := newTopic(topicType)
241
242         a, err := kafka.NewAdminClient(&kafka.ConfigMap{"bootstrap.servers": bootstrapserver})
243         defer func() { a.Close() }()
244         if err != nil {
245                 w.WriteHeader(http.StatusInternalServerError)
246                 fmt.Fprintf(w, "Cannot create client to bootstrap server: "+bootstrapserver+", error details: %v", err)
247                 return
248         }
249
250         ctx, cancel := context.WithCancel(context.Background())
251         defer cancel()
252
253         maxDur := 10 * time.Second
254
255         _, err = a.CreateTopics(
256                 ctx,
257                 []kafka.TopicSpecification{{
258                         Topic:             topicId,
259                         NumPartitions:     1,
260                         ReplicationFactor: 1}},
261                 kafka.SetAdminOperationTimeout(maxDur))
262
263         if err != nil {
264                 w.WriteHeader(http.StatusInternalServerError)
265                 fmt.Fprintf(w, "Failed to create topic: %v, error details: %v", topicId, err)
266                 return
267         }
268         topics[topicId] = t
269         w.WriteHeader(http.StatusCreated)
270         fmt.Fprintf(w, "Topic created")
271 }
272
273 // Get topic type
274 // GET on /topic/<topic>
275 func getTopic(w http.ResponseWriter, req *http.Request) {
276         t, _, exist := getTopicFromRequest(w, req)
277         if !exist {
278                 return
279         }
280         w.WriteHeader(http.StatusOK)
281         fmt.Fprintf(w, t.contentType)
282 }
283
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)
288         if !exist {
289                 return
290         }
291         ctr := mux.Vars(req)["counter"]
292
293         var ctrvalue = -1
294         if ctr == "received" {
295                 ctrvalue = int(t.counters.received.get())
296         } else if ctr == "sent" {
297                 ctrvalue = int(t.counters.sent.get())
298         }
299
300         if ctrvalue == -1 {
301                 w.WriteHeader(http.StatusBadRequest)
302                 fmt.Fprintf(w, "Counter %v does not exist", ctr)
303                 return
304         }
305         w.WriteHeader(http.StatusOK)
306         fmt.Fprintf(w, strconv.Itoa(ctrvalue))
307         return
308 }
309
310 func startToSend(w http.ResponseWriter, req *http.Request) {
311         t, topicId, exist := getTopicFromRequest(w, req)
312         fmt.Printf("Start to send to topic: %v\n", topicId)
313         if !exist {
314                 return
315         }
316
317         if !t.messages.startSend() {
318                 w.WriteHeader(http.StatusOK)
319                 fmt.Fprintf(w, "Already started sending")
320                 return
321         }
322         go func() {
323                 p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": bootstrapserver})
324                 if err != nil {
325                         fmt.Printf("Cannot create producer for topic: %v, error details: %v\n", topicId, err)
326                         return
327                 }
328                 defer func() {
329                         fmt.Printf("Closing producer for topic: %v\n", topicId)
330                         p.Close()
331                 }()
332                 for {
333                         q := t.messages.send
334                         if q == nil {
335                                 return
336                         }
337                         m, err := q.Get(0)
338                         if err == nil {
339                                 err = p.Produce(&kafka.Message{
340                                         TopicPartition: kafka.TopicPartition{Topic: &topicId, Partition: kafka.PartitionAny},
341                                         Value:          []byte(fmt.Sprintf("%v", m)),
342                                 }, nil)
343                                 if err == nil {
344                                         q.Remove(0)
345                                         t.counters.sent.step()
346                                         globalCounters.sent.step()
347                                         msg := fmt.Sprintf("%v", m)
348                                         if len(msg) < 500 {
349                                                 fmt.Printf("Message sent on topic: %v, len: %v, msg: %v\n", topicId, len(msg), msg)
350                                         } else {
351                                                 fmt.Printf("Message sent on topic: %v, len: %v, is larger than 500...not printed\n", topicId, len(msg))
352                                         }
353                                 } else {
354                                         fmt.Printf("Failed to send message on topic: %v. Discarded. Error details: %v\n", topicId, err)
355                                         q.Remove(0)
356                                 }
357                         } else {
358                                 time.Sleep(10 * time.Millisecond)
359                         }
360                 }
361         }()
362
363         w.WriteHeader(http.StatusOK)
364         fmt.Fprintf(w, "Sending started")
365 }
366
367 func startToReceive(w http.ResponseWriter, req *http.Request) {
368         t, topicId, exist := getTopicFromRequest(w, req)
369         if !exist {
370                 return
371         }
372
373         if !t.messages.startReceive() {
374                 w.WriteHeader(http.StatusOK)
375                 fmt.Fprintf(w, "Already started receiving")
376                 return
377         }
378
379         go func() {
380
381                 defer func() { t.messages.stopReceive() }()
382
383                 groudId := "kafkaprocon"
384
385                 c, err := kafka.NewConsumer(&kafka.ConfigMap{
386                         "bootstrap.servers":       bootstrapserver,
387                         "group.id":                groudId,
388                         "auto.offset.reset":       "earliest",
389                         "enable.auto.commit":      true,
390                         "auto.commit.interval.ms": 5000,
391                 })
392                 if err != nil {
393                         fmt.Printf("Cannot create consumer for topic: %v, error details: %v\n", topicId, err)
394                         t.messages.stopReceive()
395                         return
396                 }
397                 c.Commit()
398                 defer func() { c.Close() }()
399                 for {
400                         que := t.messages.received
401                         if que == nil {
402                                 fmt.Printf("Cannot start receiving on topic: %v, queue does not exist\n", topicId)
403                                 return
404                         }
405                         fmt.Printf("Start subscribing on topic: %v\n", topicId)
406                         err = c.SubscribeTopics([]string{topicId}, nil)
407                         if err != nil {
408                                 fmt.Printf("Cannot start subscribing on topic: %v, error details: %v\n", topicId, err)
409                                 return
410                         }
411                         maxDur := 1 * time.Second
412                         for {
413                                 msg, err := c.ReadMessage(maxDur)
414                                 if err == nil {
415                                         if len(msg.Value) < 500 {
416                                                 fmt.Printf("Message received on topic: %v, partion: %v, len: %v, msg: %v\n", topicId, msg.TopicPartition, len(msg.Value), string(msg.Value))
417                                         } else {
418                                                 fmt.Printf("Message received on topic: %v, partion: %v, len: %v is larger than 500...not printed\n", topicId, msg.TopicPartition, len(msg.Value))
419                                         }
420                                         err = t.messages.received.Enqueue(string(msg.Value))
421                                         if err != nil {
422                                                 w.WriteHeader(http.StatusInternalServerError)
423                                                 fmt.Fprintf(w, "Received message topic: %v, cannot be put in queue, %v", topicId, err)
424                                                 return
425                                         }
426                                         t.counters.received.step()
427                                         globalCounters.received.step()
428                                 } else {
429                                         fmt.Printf("Nothing to consume on topic: %v, reason: %v\n", topicId, err)
430                                 }
431                         }
432                 }
433         }()
434
435         w.WriteHeader(http.StatusOK)
436         fmt.Fprintf(w, "Receiving started")
437 }
438
439 // Post a message to a topic
440 // POST /send    content type is specified in content type
441 func sendToTopic(w http.ResponseWriter, req *http.Request) {
442
443         t, topicId, exist := getTopicFromRequest(w, req)
444         fmt.Printf("Send to topic: %v\n", topicId)
445         if !exist {
446                 return
447         }
448         q := t.messages.send
449         if q == nil {
450                 w.WriteHeader(http.StatusBadRequest)
451                 fmt.Fprintf(w, "Sending not initiated on topic: %v", topicId)
452                 return
453         }
454         ct := req.Header.Get("Content-Type")
455         if ct != t.contentType {
456                 w.WriteHeader(http.StatusBadRequest)
457                 fmt.Fprintf(w, "Message to send content type: %v on topic: %v does not match queue content type: %v", ct, topicId, t.contentType)
458                 return
459         }
460
461         if ct == "application/json" {
462                 // decoder := json.NewDecoder(req.Body)
463                 // var j :=
464                 // err := decoder.Decode(&j)
465                 // if err != nil {
466                 //      w.WriteHeader(http.StatusBadRequest)
467                 //      w.Header().Set("Content-Type", "text/plain")
468                 //      fmt.Fprintf(w, "Json payload cannot be decoded, error details %v\n", err)
469                 //      return
470                 // }
471                 //m = mux.Vars(req)[""]
472                 if err := req.ParseForm(); err != nil {
473                         w.WriteHeader(http.StatusBadRequest)
474                         fmt.Fprintf(w, "Json payload cannot be decoded on topic: %v, error details %v", topicId, err)
475                         return
476                 }
477                 b, err := ioutil.ReadAll(req.Body)
478                 if err == nil {
479                         if len(b) < 500 {
480                                 fmt.Printf("Json payload to send on topic: %v, msg: %v\n", topicId, string(b))
481                         } else {
482                                 fmt.Printf("Json payload to send on topic: %v larger than 500 bytes, not printed...\n", topicId)
483                         }
484                 } else {
485                         fmt.Printf("Json payload to send on topic: %v cannnot be decoded, err: %v\n", topicId, err)
486                 }
487                 err = q.Enqueue(string(b))
488                 if err != nil {
489                         w.WriteHeader(http.StatusInternalServerError)
490                         fmt.Fprintf(w, "Json message to send cannot be put in queue")
491                         return
492                 }
493         } else if ct == "text/plain" {
494                 if err := req.ParseForm(); err != nil {
495                         w.WriteHeader(http.StatusBadRequest)
496                         fmt.Fprintf(w, "Text payload to send on topic: %v cannot be decoded, error details %v\n", topicId, err)
497                         return
498                 }
499                 b, err := ioutil.ReadAll(req.Body)
500                 if err == nil {
501                         if len(b) < 500 {
502                                 fmt.Printf("Text payload to send on topic: %v, msg: %v\n", topicId, string(b))
503                         } else {
504                                 fmt.Printf("Text payload to send on topic: %v larger than 500 bytes, not printed...\n", topicId)
505                         }
506                 } else {
507                         fmt.Printf("Text payload to send on topic: %v cannnot be decoded, err: %v\n", topicId, err)
508                 }
509                 err = q.Enqueue(string(b))
510                 if err != nil {
511                         w.WriteHeader(http.StatusInternalServerError)
512                         fmt.Fprintf(w, "Text message to send cannot be put in queue")
513                         return
514                 }
515         } else {
516                 w.WriteHeader(http.StatusBadRequest)
517                 fmt.Fprintf(w, "Message to send, unknown content type %v", ct)
518                 return
519         }
520
521         w.WriteHeader(http.StatusOK)
522         w.Header().Set("Content-Type", "text/plain")
523         fmt.Fprintf(w, "Message to send put in queue")
524 }
525
526 // Get zero or one message from a topic
527 // GET /receive
528 func receiveFromTopic(w http.ResponseWriter, req *http.Request) {
529         t, topicId, exist := getTopicFromRequest(w, req)
530         if !exist {
531                 return
532         }
533         if t.messages.received == nil {
534                 w.WriteHeader(http.StatusBadRequest)
535                 fmt.Fprintf(w, "Receiving not initiated on topic %v", topicId)
536                 return
537         }
538
539         m, err := t.messages.received.Dequeue()
540         if err != nil {
541                 w.WriteHeader(http.StatusNoContent)
542                 return
543         }
544
545         w.Header().Set("Content-Type", t.contentType)
546         w.WriteHeader(http.StatusOK)
547         fmt.Fprintf(w, "%v", m)
548 }
549
550 // Remove the send queue to stop sending
551 func stopToSend(w http.ResponseWriter, req *http.Request) {
552         fmt.Printf("Stop sending\n")
553         t, _, exist := getTopicFromRequest(w, req)
554         if !exist {
555                 return
556         }
557         t.messages.stopSend()
558         w.WriteHeader(http.StatusNoContent)
559 }
560
561 // Remove the receive queue to stop receiving
562 func stopToReceive(w http.ResponseWriter, req *http.Request) {
563         fmt.Printf("Stop receiving\n")
564         t, _, exist := getTopicFromRequest(w, req)
565         if !exist {
566                 return
567         }
568         t.messages.stopReceive()
569         w.WriteHeader(http.StatusNoContent)
570 }
571
572 func HelloServer(w http.ResponseWriter, r *http.Request) {
573         fmt.Fprintf(w, "Hello, %s!", r.URL.Path[1:])
574 }
575
576 func main() {
577
578         initApp()
579
580         r := mux.NewRouter()
581
582         r.HandleFunc("/", healthCheck).Methods("GET")
583         r.HandleFunc("/reset", allreset).Methods("POST")
584         r.HandleFunc("/counters/{counter}", getCounter).Methods("GET")
585         r.HandleFunc("/topics", getTopics).Methods("GET")
586         r.HandleFunc("/topics/{topic}/counters/{counter}", getTopicCounter).Methods("GET")
587         r.HandleFunc("/topics/{topic}", createTopic).Methods("PUT").Queries("type", "{type}")
588         r.HandleFunc("/topics/{topic}", getTopic).Methods("GET")
589         r.HandleFunc("/topics/{topic}/startsend", startToSend).Methods("POST")
590         r.HandleFunc("/topics/{topic}/startreceive", startToReceive).Methods("POST")
591         r.HandleFunc("/topics/{topic}/stopsend", stopToSend).Methods("POST")
592         r.HandleFunc("/topics/{topic}/stopreceive", stopToReceive).Methods("POST")
593         r.HandleFunc("/topics/{topic}/msg", sendToTopic).Methods("POST")
594         r.HandleFunc("/topics/{topic}/msg", receiveFromTopic).Methods("GET")
595
596         port := "8090"
597         srv := &http.Server{
598                 Handler:      r,
599                 Addr:         ":" + port,
600                 WriteTimeout: 15 * time.Second,
601                 ReadTimeout:  15 * time.Second,
602         }
603         fmt.Println("Running on port: " + port)
604         fmt.Printf(srv.ListenAndServe().Error())
605 }