(kafkaWriter *kafka.Writer)
| 11 | ) |
| 12 | |
| 13 | func producerHandler(kafkaWriter *kafka.Writer) func(http.ResponseWriter, *http.Request) { |
| 14 | return http.HandlerFunc(func(wrt http.ResponseWriter, req *http.Request) { |
| 15 | body, err := ioutil.ReadAll(req.Body) |
| 16 | if err != nil { |
| 17 | log.Fatalln(err) |
| 18 | } |
| 19 | msg := kafka.Message{ |
| 20 | Key: []byte(fmt.Sprintf("address-%s", req.RemoteAddr)), |
| 21 | Value: body, |
| 22 | } |
| 23 | err = kafkaWriter.WriteMessages(req.Context(), msg) |
| 24 | |
| 25 | if err != nil { |
| 26 | wrt.Write([]byte(err.Error())) |
| 27 | log.Fatalln(err) |
| 28 | } |
| 29 | }) |
| 30 | } |
| 31 | |
| 32 | func getKafkaWriter(kafkaURL, topic string) *kafka.Writer { |
| 33 | return &kafka.Writer{ |
no test coverage detected