Merge pull request #572 from marle3003/develop

Develop

Author: marle3003

README.md

@@ -35,6 +35,10 @@ the risk of bugs in production.
 - **Multiple Provider support**: File, HTTP, GIT, NPM to gather configurations and scripts.
 - **Dashboard** to see what's going on.
 
+## 🧩 OpenAPI & AsyncAPI Support
+
+Mokapi supports mocking REST APIs using **OpenAPI 2.0 / 3.0 / 3.1** and event-driven systems using **AsyncAPI**.
+
 ## 🔧 Spin Up Mokapi
 
 Install and start Mokapi using one of the following methods.

acceptance/petstore_test.go

@@ -60,6 +60,7 @@ func (suite *PetStoreSuite) TestApi() {
 					"description": "",
 					"host":        "127.0.0.1:19092",
 					"name":        "broker",
+					"protocol":    "kafka",
 				},
 			},
 

api/handler_kafka.go

@@ -31,6 +31,7 @@ type kafka struct {
 type kafkaServer struct {
 	Name        string           `json:"name"`
 	Host        string           `json:"host"`
+	Protocol    string           `json:"protocol"`
 	Description string           `json:"description"`
 	Tags        []kafkaServerTag `json:"tags,omitempty"`
 }
@@ -173,6 +174,7 @@ func getKafka(info *runtime.KafkaInfo) kafka {
 			Name:        name,
 			Host:        s.Value.Host,
 			Description: s.Value.Description,
+			Protocol:    s.Value.Protocol,
 		}
 		for _, r := range s.Value.Tags {
 			if r.Value == nil {

api/handler_kafka_test.go

@@ -103,7 +103,7 @@ func TestHandler_Kafka(t *testing.T) {
 				}))
 			},
 			requestUrl:   "http://foo.api/api/services/kafka/foo",
-			responseBody: `{"name":"foo","description":"bar","version":"1.0","servers":[{"name":"foo","host":"foo.bar","description":"bar"}]}`,
+			responseBody: `{"name":"foo","description":"bar","version":"1.0","servers":[{"name":"foo","host":"foo.bar","protocol":"kafka","description":"bar"}]}`,
 		},
 		{
 			name: "server with tags",
@@ -122,7 +122,7 @@ func TestHandler_Kafka(t *testing.T) {
 				}))
 			},
 			requestUrl:   "http://foo.api/api/services/kafka/foo",
-			responseBody: `{"name":"foo","description":"bar","version":"1.0","servers":[{"name":"foo","host":"foo.bar","description":"bar","tags":[{"name":"env:test","description":"This environment is for running internal tests"}]}]}`,
+			responseBody: `{"name":"foo","description":"bar","version":"1.0","servers":[{"name":"foo","host":"foo.bar","protocol":"kafka","description":"bar","tags":[{"name":"env:test","description":"This environment is for running internal tests"}]}]}`,
 		},
 		{
 			name: "get specific with topic",
@@ -193,7 +193,7 @@ func TestHandler_Kafka(t *testing.T) {
 				return app
 			},
 			requestUrl:   "http://foo.api/api/services/kafka/foo",
-			responseBody: `{"name":"foo","description":"bar","version":"1.0","servers":[{"name":"foo","host":"foo.bar","description":""}],"groups":[{"name":"foo","members":null,"coordinator":"foo.bar:9092","leader":"","state":"PreparingRebalance","protocol":"range","topics":null}]}`,
+			responseBody: `{"name":"foo","description":"bar","version":"1.0","servers":[{"name":"foo","host":"foo.bar","protocol":"kafka","description":""}],"groups":[{"name":"foo","members":null,"coordinator":"foo.bar:9092","leader":"","state":"PreparingRebalance","protocol":"range","topics":null}]}`,
 		},
 		{
 			name: "get specific with group containing members",
@@ -249,7 +249,7 @@ func TestHandler_Kafka(t *testing.T) {
 				return app
 			},
 			requestUrl:   "http://foo.api/api/services/kafka/foo",
-			responseBody: `{"name":"foo","description":"bar","version":"1.0","servers":[{"name":"foo","host":"foo.bar","description":""}],"groups":[{"name":"foo","members":[{"name":"m1","addr":"192.168.0.100","clientSoftwareName":"mokapi","clientSoftwareVersion":"1.0","heartbeat":"2024-04-22T15:04:05+07:00","partitions":{"topic":[1,2,5]}},{"name":"m2","addr":"192.168.0.200","clientSoftwareName":"mokapi","clientSoftwareVersion":"1.0","heartbeat":"2024-04-22T15:04:10+07:00","partitions":{"topic":[3,4,6]}}],"coordinator":"foo.bar:9092","leader":"m1","state":"PreparingRebalance","protocol":"range","topics":null}]}`,
+			responseBody: `{"name":"foo","description":"bar","version":"1.0","servers":[{"name":"foo","host":"foo.bar","protocol":"kafka","description":""}],"groups":[{"name":"foo","members":[{"name":"m1","addr":"192.168.0.100","clientSoftwareName":"mokapi","clientSoftwareVersion":"1.0","heartbeat":"2024-04-22T15:04:05+07:00","partitions":{"topic":[1,2,5]}},{"name":"m2","addr":"192.168.0.200","clientSoftwareName":"mokapi","clientSoftwareVersion":"1.0","heartbeat":"2024-04-22T15:04:10+07:00","partitions":{"topic":[3,4,6]}}],"coordinator":"foo.bar:9092","leader":"m1","state":"PreparingRebalance","protocol":"range","topics":null}]}`,
 		},
 		{
 			name: "get specific with topic and openapi schema",

buffer/page.go

@@ -0,0 +1,143 @@
+package buffer
+
+import (
+	"io"
+	"sync"
+)
+
+const pageSize = 65536
+
+var (
+	pagePool = sync.Pool{New: func() interface{} { return new(page) }}
+)
+
+type page struct {
+	offset int // absolute offset
+	buffer [pageSize]byte
+	length int // length of this page
+	refs   refCounter
+}
+
+type fragment struct {
+	pages  []*page
+	offset int
+	cursor int
+	length int
+}
+
+func newPage(offset int) *page {
+	p := pagePool.Get().(*page)
+	p.offset = offset
+	p.refs.inc()
+	return p
+}
+
+func (p *page) unref() {
+	if p.refs.dec() {
+		p.offset = 0
+		p.length = 0
+		pagePool.Put(p)
+	}
+}
+
+func (p *page) ReadAt(b []byte, offset int) (int, error) {
+	if offset -= p.offset; offset < 0 || offset > pageSize {
+		panic("offset out of range")
+	}
+
+	if offset > p.length {
+		return 0, nil
+	}
+
+	n := copy(b, p.buffer[offset:p.length])
+	return n, nil
+}
+
+func (p *page) Write(b []byte) (n int, err error) {
+	n = copy(p.buffer[p.length:], b)
+	p.length += n
+	return
+}
+
+func (p *page) WriteAt(b []byte, offset int) (int, error) {
+	if offset -= p.offset; offset < 0 || offset > pageSize {
+		panic("offset out of range")
+	}
+	n := copy(p.buffer[offset:], b)
+	if end := offset + n; end > p.length {
+		p.length += end
+	}
+	return n, nil
+}
+
+func (p *page) Size() int {
+	return p.length
+}
+
+func (p *page) slice(begin, end int) []byte {
+	i, j := begin-p.offset, end-p.offset
+	if i < 0 {
+		i = 0
+	}
+	if j > len(p.buffer) {
+		j = len(p.buffer)
+	}
+	return p.buffer[i:j]
+}
+
+func (f *fragment) Read(b []byte) (int, error) {
+	if end := f.offset + f.length; f.cursor >= end {
+		return 0, io.EOF
+	}
+	if len(b) > f.length {
+		b = b[:f.length]
+	}
+	read := 0
+	for _, p := range f.pages {
+		n, _ := p.ReadAt(b, f.cursor)
+		b = b[n:]
+		f.cursor += n
+		read += n
+	}
+	if read == 0 {
+		return 0, io.EOF
+	}
+	return read, nil
+}
+
+func (f *fragment) Size() int {
+	return f.length
+}
+
+func (f *fragment) Seek(offset int64, whence int) (int64, error) {
+	switch whence {
+	case io.SeekCurrent:
+		f.cursor += int(offset)
+	case io.SeekStart:
+		f.cursor = f.offset + int(offset)
+	case io.SeekEnd:
+		f.cursor = f.offset + f.length + int(offset)
+	}
+	return int64(f.cursor), nil
+}
+
+func (f *fragment) ref() {
+	for _, p := range f.pages {
+		p.refs.inc()
+	}
+}
+
+func (f *fragment) unref() {
+	for _, p := range f.pages {
+		p.unref()
+	}
+	f.pages = nil
+	f.length = 0
+	f.cursor = 0
+	f.offset = 0
+}
+
+func (f *fragment) Close() error {
+	f.unref()
+	return nil
+}

buffer/pagebuffer.go

@@ -0,0 +1,156 @@
+package buffer
+
+import (
+	"encoding/binary"
+	"io"
+	"sync"
+	"sync/atomic"
+)
+
+var (
+	pageBufferPool = sync.Pool{New: func() interface{} { return new(pageBuffer) }}
+)
+
+type refCounter uint32
+
+type pageBuffer struct {
+	pages  []*page
+	length int
+	cursor int
+	refs   refCounter
+}
+
+type Writer interface {
+}
+
+type Buffer interface {
+	io.Writer
+	WriteTo(io.Writer) (int, error)
+	WriteAt([]byte, int)
+	WriteSizeAt(size int, offset int)
+	Scan(begin, end int, f func([]byte) bool)
+	Size() int
+	Slice(begin, end int) Fragment
+	Unref()
+}
+
+type Fragment interface {
+	io.ReadCloser
+	io.Seeker
+	Size() int
+}
+
+func (r *refCounter) inc() {
+	atomic.AddUint32((*uint32)(r), 1)
+}
+
+func (r *refCounter) dec() bool {
+	i := atomic.AddUint32((*uint32)(r), ^uint32(0))
+	return i == 0
+}
+
+func NewPageBuffer() Buffer {
+	pb := pageBufferPool.Get().(*pageBuffer)
+	pb.refs.inc()
+	return pb
+}
+
+func (pb *pageBuffer) Unref() {
+	if pb.refs.dec() {
+		for _, p := range pb.pages {
+			p.unref()
+		}
+		pb.length = 0
+		pb.cursor = 0
+		pb.pages = nil
+		pageBufferPool.Put(pb)
+	}
+}
+
+func (pb *pageBuffer) Write(b []byte) (n int, err error) {
+	n = len(b)
+	if len(pb.pages) == 0 {
+		pb.addPage()
+	}
+
+	for len(b) != 0 {
+		tail := pb.pages[len(pb.pages)-1]
+		available := pageSize - tail.Size()
+
+		if len(b) <= available {
+			tail.Write(b)
+			pb.length += len(b)
+			break
+		}
+
+		tail.Write(b[:available])
+		b = b[available:]
+		pb.length += available
+		pb.addPage()
+	}
+
+	return
+}
+
+func (pb *pageBuffer) WriteSizeAt(size int, offset int) {
+	var b [4]byte
+	binary.BigEndian.PutUint32(b[:], uint32(size))
+	pb.WriteAt(b[:], offset)
+}
+
+func (pb *pageBuffer) WriteAt(b []byte, offset int) {
+	for _, p := range pb.slice(offset, offset+len(b)) {
+		n, _ := p.WriteAt(b, offset)
+		b = b[n:]
+		offset += n
+	}
+}
+
+func (pb *pageBuffer) Size() int {
+	return pb.length
+}
+
+func (pb *pageBuffer) addPage() {
+	p := newPage(pb.length)
+	pb.pages = append(pb.pages, p)
+}
+
+func (pb *pageBuffer) slice(begin, end int) []*page {
+	i := begin / pageSize
+	j := end / pageSize
+	if j < len(pb.pages) {
+		j++
+	}
+	return pb.pages[i:j]
+}
+
+func (pb *pageBuffer) Scan(begin, end int, f func([]byte) bool) {
+	for _, p := range pb.slice(begin, end) {
+		if !f(p.slice(begin, end)) {
+			return
+		}
+	}
+}
+
+func (pb *pageBuffer) WriteTo(w io.Writer) (written int, err error) {
+	pb.Scan(pb.cursor, pb.length, func(b []byte) bool {
+		var n int
+		n, err = w.Write(b)
+		written += n
+		return err == nil
+	})
+	pb.cursor += written
+	return
+}
+
+func (pb *pageBuffer) Slice(begin, end int) Fragment {
+	pages := pb.slice(begin, end)
+	f := &fragment{
+		pages:  pages,
+		offset: begin,
+		cursor: begin,
+		length: end - begin,
+	}
+	f.ref()
+	return f
+}

buffer/pagebuffer_test.go

@@ -0,0 +1 @@
+package buffer

cmd/mokapi/main.go

@@ -113,11 +113,13 @@ func createServer(cfg *static.Config) (*server.Server, error) {
 	}
 	http := server.NewHttpManager(scriptEngine, certStore, app)
 	kafka := server.NewKafkaManager(scriptEngine, app)
+	mqtt := server.NewMqttManager(scriptEngine, app)
 	smtp := server.NewSmtpManager(app, scriptEngine, certStore)
 	ldap := server.NewLdapDirectoryManager(scriptEngine, certStore, app)
 
 	watcher.AddListener(func(e dynamic.ConfigEvent) {
 		kafka.UpdateConfig(e)
+		mqtt.UpdateConfig(e)
 		http.Update(e)
 		smtp.UpdateConfig(e)
 		ldap.UpdateConfig(e)

config/dynamic/asyncApi/convert_test.go

@@ -38,7 +38,7 @@ func TestConfig_Convert(t *testing.T) {
 	// Server
 	require.Len(t, cfg3.Servers, 1)
 	require.Equal(t, "test.mosquitto.org:{port}", cfg3.Servers["production"].Value.Host)
-	require.Equal(t, "mqtt", cfg3.Servers["production"].Value.Protocol)
+	require.Equal(t, "kafka", cfg3.Servers["production"].Value.Protocol)
 	require.Equal(t, "Test broker", cfg3.Servers["production"].Value.Description)
 
 	// Channel