Develop

api/handler.go

@@ -1,9 +1,9 @@
 package api
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
-	log "github.com/sirupsen/logrus"
 	"io/fs"
 	"mokapi/config/static"
 	"mokapi/runtime"
@@ -15,6 +15,8 @@ import (
 	"slices"
 	"strconv"
 	"strings"
+
+	log "github.com/sirupsen/logrus"
 )
 
 type handler struct {
@@ -220,12 +222,18 @@ func (h *handler) getInfo(w http.ResponseWriter, _ *http.Request) {
 	writeJsonBody(w, i)
 }
 
-func writeJsonBody(w http.ResponseWriter, i interface{}) {
-	b, err := json.Marshal(i)
+func writeJsonBody(w http.ResponseWriter, v interface{}) {
+	var buf bytes.Buffer
+	enc := json.NewEncoder(&buf)
+	enc.SetEscapeHTML(false)
+	err := enc.Encode(v) // includes newline
+
 	if err != nil {
 		writeError(w, err, http.StatusInternalServerError)
 		return
 	}
+
+	b := bytes.TrimSuffix(buf.Bytes(), []byte("\n"))
 	_, err = w.Write(b)
 	if err != nil {
 		log.Errorf("write response body failed: %v", err)

api/handler_config_test.go

@@ -248,7 +248,7 @@ func TestHandler_Config(t *testing.T) {
 			test: []try.ResponseCondition{
 				try.HasStatusCode(http.StatusOK),
 				try.HasHeader("Content-Type", "application/json"),
-				try.HasBody(`{"id":"61373430-3061-3131-6663-326332386638","url":"https://git.bar?file=/foo/foo.json\u0026ref=main","provider":"git","time":"2023-12-27T13:01:30Z"}`),
+				try.HasBody(`{"id":"61373430-3061-3131-6663-326332386638","url":"https://git.bar?file=/foo/foo.json&ref=main","provider":"git","time":"2023-12-27T13:01:30Z"}`),
 			},
 		},
 		{

api/handler_events_test.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"mokapi/config/static"
+	"mokapi/providers/asyncapi3/kafka/store"
 	"mokapi/providers/openapi"
 	"mokapi/runtime"
 	"mokapi/runtime/events"
@@ -208,3 +209,127 @@ func TestHandler_Events(t *testing.T) {
 		})
 	}
 }
+
+func TestHandler_KafkaEvents(t *testing.T) {
+	testcases := []struct {
+		name string
+		fn   func(t *testing.T, h http.Handler, sm *events.StoreManager)
+	}{
+		{
+			name: "empty kafka events",
+			fn: func(t *testing.T, h http.Handler, sm *events.StoreManager) {
+				try.Handler(t,
+					http.MethodGet,
+					"http://foo.api/api/events?namespace=kafka",
+					nil,
+					"",
+					h,
+					try.HasStatusCode(200),
+					try.HasHeader("Content-Type", "application/json"),
+					try.HasBody(`[]`))
+			},
+		},
+		{
+			name: "with kafka events",
+			fn: func(t *testing.T, h http.Handler, sm *events.StoreManager) {
+				sm.SetStore(1, events.NewTraits().WithNamespace("kafka"))
+				err := sm.Push(&eventstest.Event{Name: "foo"}, events.NewTraits().WithNamespace("kafka"))
+				event := sm.GetEvents(events.NewTraits())[0]
+				require.NoError(t, err)
+				try.Handler(t,
+					http.MethodGet,
+					"http://foo.api/api/events?namespace=kafka",
+					nil,
+					"",
+					h,
+					try.HasStatusCode(200),
+					try.HasHeader("Content-Type", "application/json"),
+					try.HasBody(fmt.Sprintf(`[{"id":"%v","traits":{"namespace":"kafka"},"data":{"Name":"foo","api":""},"time":"%v"}]`,
+						event.Id,
+						event.Time.Format(time.RFC3339Nano))))
+			},
+		},
+		{
+			name: "get specific event",
+			fn: func(t *testing.T, h http.Handler, sm *events.StoreManager) {
+				sm.SetStore(1, events.NewTraits().WithNamespace("kafka"))
+				err := sm.Push(&eventstest.Event{Name: "foo"}, events.NewTraits().WithNamespace("kafka"))
+				event := sm.GetEvents(events.NewTraits())[0]
+				require.NoError(t, err)
+				try.Handler(t,
+					http.MethodGet,
+					"http://foo.api/api/events/"+event.Id,
+					nil,
+					"",
+					h,
+					try.HasStatusCode(200),
+					try.HasHeader("Content-Type", "application/json"),
+					try.HasBody(fmt.Sprintf(`{"id":"%v","traits":{"namespace":"kafka"},"data":{"Name":"foo","api":""},"time":"%v"}`,
+						event.Id,
+						event.Time.Format(time.RFC3339Nano))))
+			},
+		},
+		{
+			name: "get kafka with producerId",
+			fn: func(t *testing.T, h http.Handler, sm *events.StoreManager) {
+				sm.SetStore(1, events.NewTraits().WithNamespace("kafka"))
+
+				err := sm.Push(&store.KafkaLog{
+					Offset:         123,
+					Key:            store.LogValue{},
+					Message:        store.LogValue{},
+					MessageId:      "foo-1",
+					Partition:      1,
+					ProducerId:     3,
+					ProducerEpoch:  1,
+					SequenceNumber: 2,
+				}, events.NewTraits().WithNamespace("kafka"))
+				require.NoError(t, err)
+
+				try.Handler(t,
+					http.MethodGet,
+					"http://foo.api/api/events?namespace=kafka",
+					nil,
+					"",
+					h,
+					try.HasStatusCode(200),
+					try.AssertBody(func(t *testing.T, body string) {
+						var m []map[string]any
+						require.NoError(t, json.Unmarshal([]byte(body), &m))
+						require.Equal(t, map[string]any{
+							"api":     "",
+							"deleted": false,
+							"headers": interface{}(nil),
+							"key": map[string]interface{}{
+								"binary": interface{}(nil),
+								"value":  "",
+							},
+							"message": map[string]interface{}{
+								"binary": interface{}(nil),
+								"value":  "",
+							},
+							"messageId":      "foo-1",
+							"offset":         float64(123),
+							"partition":      float64(1),
+							"producerEpoch":  float64(1),
+							"producerId":     float64(3),
+							"schemaId":       float64(0),
+							"sequenceNumber": float64(2),
+						},
+							m[0]["data"])
+					}))
+			},
+		},
+	}
+
+	for _, tc := range testcases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			cfg := &static.Config{}
+			app := runtime.New(cfg)
+
+			h := New(app, static.Api{})
+			tc.fn(t, h, app.Events)
+		})
+	}
+}

api/handler_kafka.go

@@ -120,11 +120,11 @@ type produceRequest struct {
 	Records []store.Record `json:"records"`
 }
 
-type produceResponse struct {
-	Offsets []recordResult `json:"offsets"`
+type ProduceResponse struct {
+	Offsets []RecordResult `json:"offsets"`
 }
 
-type recordResult struct {
+type RecordResult struct {
 	Partition int
 	Offset    int64
 	Error     string
@@ -222,9 +222,9 @@ func (h *handler) handleKafka(w http.ResponseWriter, r *http.Request) {
 					writeError(w, err, http.StatusBadRequest)
 				}
 			}
-			res := produceResponse{}
+			res := ProduceResponse{}
 			for _, rec := range result {
-				res.Offsets = append(res.Offsets, recordResult{
+				res.Offsets = append(res.Offsets, RecordResult{
 					Partition: rec.Partition,
 					Offset:    rec.Offset,
 					Error:     rec.Error,
@@ -247,7 +247,7 @@ func (h *handler) handleKafka(w http.ResponseWriter, r *http.Request) {
 			w.WriteHeader(http.StatusNotFound)
 		} else {
 			w.Header().Set("Content-Type", "application/json")
-			writeJsonBody(w, getPartitions(k, t))
+			writeJsonBody(w, getPartitions(t))
 		}
 		return
 	// /api/services/kafka/{cluster}/topics/{topic}/partitions/{id}
@@ -276,7 +276,7 @@ func (h *handler) handleKafka(w http.ResponseWriter, r *http.Request) {
 		}
 		if r.Method == "GET" {
 			w.Header().Set("Content-Type", "application/json")
-			writeJsonBody(w, newPartition(k.Store, p))
+			writeJsonBody(w, newPartition(p))
 		} else {
 			records, err := getProduceRecords(r)
 			if err != nil {
@@ -296,9 +296,9 @@ func (h *handler) handleKafka(w http.ResponseWriter, r *http.Request) {
 					writeError(w, err, http.StatusBadRequest)
 				}
 			}
-			res := produceResponse{}
+			res := ProduceResponse{}
 			for _, rec := range result {
-				res.Offsets = append(res.Offsets, recordResult{
+				res.Offsets = append(res.Offsets, RecordResult{
 					Partition: rec.Partition,
 					Offset:    rec.Offset,
 					Error:     rec.Error,
@@ -458,7 +458,7 @@ func getTopics(info *runtime.KafkaInfo) []topic {
 			addr = name
 		}
 		t := info.Store.Topic(addr)
-		topics = append(topics, newTopic(info.Store, t, ch.Value, info.Config))
+		topics = append(topics, newTopic(t, ch.Value, info.Config))
 	}
 	sort.Slice(topics, func(i, j int) bool {
 		return strings.Compare(topics[i].Name, topics[j].Name) < 0
@@ -477,18 +477,18 @@ func getTopic(info *runtime.KafkaInfo, name string) *topic {
 		}
 		if addr == name {
 			t := info.Store.Topic(addr)
-			r := newTopic(info.Store, t, ch.Value, info.Config)
+			r := newTopic(t, ch.Value, info.Config)
 			return &r
 		}
 
 	}
 	return nil
 }
 
-func newTopic(s *store.Store, t *store.Topic, ch *asyncapi3.Channel, cfg *asyncapi3.Config) topic {
+func newTopic(t *store.Topic, ch *asyncapi3.Channel, cfg *asyncapi3.Config) topic {
 	var partitions []partition
 	for _, p := range t.Partitions {
-		partitions = append(partitions, newPartition(s, p))
+		partitions = append(partitions, newPartition(p))
 	}
 	sort.Slice(partitions, func(i, j int) bool {
 		return partitions[i].Id < partitions[j].Id
@@ -549,10 +549,10 @@ func newTopic(s *store.Store, t *store.Topic, ch *asyncapi3.Channel, cfg *asynca
 	return result
 }
 
-func getPartitions(info *runtime.KafkaInfo, t *store.Topic) []partition {
+func getPartitions(t *store.Topic) []partition {
 	var partitions []partition
 	for _, p := range t.Partitions {
-		partitions = append(partitions, newPartition(info.Store, p))
+		partitions = append(partitions, newPartition(p))
 	}
 	sort.Slice(partitions, func(i, j int) bool {
 		return partitions[i].Id < partitions[j].Id
@@ -594,13 +594,12 @@ func newGroup(g *store.Group) group {
 	return grp
 }
 
-func newPartition(s *store.Store, p *store.Partition) partition {
-	leader, _ := s.Broker(p.Leader)
+func newPartition(p *store.Partition) partition {
 	return partition{
 		Id:          p.Index,
 		StartOffset: p.StartOffset(),
 		Offset:      p.Offset(),
-		Leader:      newBroker(leader),
+		Leader:      newBroker(p.Leader),
 		Segments:    len(p.Segments),
 	}
 }
@@ -649,7 +648,7 @@ func getProduceRecords(r *http.Request) ([]store.Record, error) {
 	return pr.Records, nil
 }
 
-func (r *recordResult) MarshalJSON() ([]byte, error) {
+func (r *RecordResult) MarshalJSON() ([]byte, error) {
 	aux := &struct {
 		Partition int     `json:"partition"`
 		Offset    int64   `json:"offset"`