Skip to content

Commit acca88b

Browse files
shohamroditimemphisshohamroditimemphis
committed
fix issues
1 parent 842b632 commit acca88b

File tree

8 files changed

+34
-48
lines changed

8 files changed

+34
-48
lines changed

‎.vscode/launch.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@
1212
"REFRESH_JWT_SECRET": "35c7b3eb969db065bfa3c66b38e4323e8f73113f3965dfb55c6bc585dcb0ba62bd399e2588fdc8f709ae0b63fb24be32590f134506ca1d7a4314339f11b8045a",
1313
"CONNECTION_TOKEN": "memphis",
1414
"ROOT_USER": "root",
15-
"HTTP_HOST":"localhost",
15+
"MEMPHIS_HOST":"localhost",
1616
"HTTP_PORT": ":3000"
1717
}
1818
}

‎conf/config.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ type Configuration struct {
1212
HTTP_PORT string
1313
ROOT_USER string
1414
CONNECTION_TOKEN string
15-
HTTP_HOST string
15+
MEMPHIS_HOST string
1616
}
1717

1818
func GetConfig() Configuration {

‎conf/config.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
{
22
"HTTP_PORT": ":3000",
3-
"HTTP_HOST": "localhost",
3+
"MEMPHIS_HOST": "localhost",
44
"ROOT_USER": "root",
55
"CONNECTION_TOKEN": "memphis",
66
"JWT_EXPIRES_IN_MINUTES": 15,

‎go.mod‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,10 +5,12 @@ go 1.18
55
require (
66
github.com/gofiber/fiber/v2 v2.40.1
77
github.com/memphisdev/memphis.go v0.1.6
8+
github.com/tkanos/gonfig v0.0.0-20210106201359-53e13348de2f
89
)
910

1011
require (
1112
github.com/andybalholm/brotli v1.0.4 // indirect
13+
github.com/ghodss/yaml v1.0.0 // indirect
1214
github.com/klauspost/compress v1.15.9 // indirect
1315
github.com/mattn/go-colorable v0.1.13 // indirect
1416
github.com/mattn/go-isatty v0.0.16 // indirect
@@ -26,4 +28,5 @@ require (
2628
golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab // indirect
2729
golang.org/x/time v0.2.0 // indirect
2830
google.golang.org/protobuf v1.28.1 // indirect
31+
gopkg.in/yaml.v2 v2.4.0 // indirect
2932
)

‎go.sum‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
github.com/andybalholm/brotli v1.0.4 h1:V7DdXeJtZscaqfNuAdSRuRFzuiKlHSC/Zh3zl9qY3JY=
22
github.com/andybalholm/brotli v1.0.4/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig=
3+
github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk=
4+
github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
35
github.com/gofiber/fiber/v2 v2.40.1 h1:pc7n9VVpGIqNsvg9IPLQhyFEMJL8gCs1kneH5D1pIl4=
46
github.com/gofiber/fiber/v2 v2.40.1/go.mod h1:Gko04sLksnHbzLSRBFWPFdzM9Ws9pRxvvIaohJK1dsk=
57
github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
@@ -29,6 +31,8 @@ github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
2931
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
3032
github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY=
3133
github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
34+
github.com/tkanos/gonfig v0.0.0-20210106201359-53e13348de2f h1:xDFq4NVQD34ekH5UsedBSgfxsBuPU2aZf7v4t0tH2jY=
35+
github.com/tkanos/gonfig v0.0.0-20210106201359-53e13348de2f/go.mod h1:DaZPBuToMc2eezA9R9nDAnmS2RMwL7yEa5YD36ESQdI=
3236
github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
3337
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
3438
github.com/valyala/fasthttp v1.41.0 h1:zeR0Z1my1wDHTRiamBCXVglQdbUwgb9uWG3k1HQz6jY=
@@ -62,3 +66,5 @@ golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1N
6266
google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
6367
google.golang.org/protobuf v1.28.1 h1:d0NfwRgPtno5B1Wa6L2DAG+KivqkdutMf1UhdNx175w=
6468
google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
69+
gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
70+
gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=

‎handlers/producer.go‎

Lines changed: 19 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,15 @@
11
package handler
22

33
import (
4-
"encoding/json"
54
"errors"
65
"strings"
76

87
"github.com/gofiber/fiber/v2"
98
"github.com/memphisdev/memphis.go"
109
)
1110

11+
var producers = make(map[string]*memphis.Producer)
12+
1213
func handleHeaders(headers map[string]string) (memphis.Headers, error) {
1314
hdrs := memphis.Headers{}
1415
hdrs.New()
@@ -22,44 +23,31 @@ func handleHeaders(headers map[string]string) (memphis.Headers, error) {
2223
return hdrs, nil
2324
}
2425

25-
func handleJsonMessage(bodyReq []byte, headers map[string]string) ([]byte, memphis.Headers, error) {
26-
type body struct {
27-
Message string `json:"message"`
28-
}
29-
var bodyRequest body
30-
err := json.Unmarshal(bodyReq, &bodyRequest)
31-
if err != nil {
32-
return nil, memphis.Headers{}, err
33-
}
34-
35-
hdrs, err := handleHeaders(headers)
36-
if err != nil {
37-
return nil, memphis.Headers{}, err
26+
func createProducer(conn *memphis.Conn, producers map[string]*memphis.Producer, stationName string) (*memphis.Producer, error) {
27+
producerName := "http_proxy"
28+
var producer *memphis.Producer
29+
var err error
30+
if _, ok := producers[stationName]; !ok {
31+
producer, err = conn.CreateProducer(stationName, producerName, memphis.ProducerGenUniqueSuffix())
32+
if err != nil {
33+
return nil, err
34+
}
35+
producers[stationName] = producer
36+
} else {
37+
producer = producers[stationName]
3838
}
3939

40-
message, err := json.Marshal(bodyRequest.Message)
41-
if err != nil {
42-
return nil, memphis.Headers{}, err
43-
}
44-
return message, hdrs, nil
40+
return producer, nil
4541
}
4642

4743
func CreateHandleMessage(conn *memphis.Conn) func(*fiber.Ctx) error {
48-
producers := make(map[string]*memphis.Producer)
4944
return func(c *fiber.Ctx) error {
5045
stationName := c.Params("stationName")
51-
producerName := "http_proxy"
5246
var producer *memphis.Producer
53-
var err error
5447

55-
if _, ok := producers[stationName]; !ok {
56-
producer, err = conn.CreateProducer(stationName, producerName, memphis.ProducerGenUniqueSuffix())
57-
if err != nil {
58-
return err
59-
}
60-
producers[stationName] = producer
61-
} else {
62-
producer = producers[stationName]
48+
producer, err := createProducer(conn, producers, stationName)
49+
if err != nil {
50+
return err
6351
}
6452

6553
bodyReq := c.Body()
@@ -73,18 +61,7 @@ func CreateHandleMessage(conn *memphis.Conn) func(*fiber.Ctx) error {
7361
}
7462

7563
switch contentType {
76-
case "application/json":
77-
message, hdrs, err = handleJsonMessage(bodyReq, headers)
78-
if err != nil {
79-
return err
80-
}
81-
case "text/":
82-
message = bodyReq
83-
hdrs, err = handleHeaders(headers)
84-
if err != nil {
85-
return err
86-
}
87-
case "application/x-protobuf":
64+
case "application/json", "text/", "application/x-protobuf":
8865
message = bodyReq
8966
hdrs, err = handleHeaders(headers)
9067
if err != nil {

‎main.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ func main() {
1717
app.Use(cors.New())
1818

1919
configuration := conf.GetConfig()
20-
conn, err := memphis.Connect(configuration.HTTP_HOST, configuration.ROOT_USER, configuration.CONNECTION_TOKEN)
20+
conn, err := memphis.Connect(configuration.MEMPHIS_HOST, configuration.ROOT_USER, configuration.CONNECTION_TOKEN)
2121
if err != nil {
2222
fmt.Println(err.Error())
2323
os.Exit(1)

‎router/router.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,6 @@ import (
99

1010
// SetupRoutes setup router api
1111
func SetupRoutes(app *fiber.App, conn *memphis.Conn) {
12-
api := app.Group("/", logger.New())
13-
api.Post("/stations/:stationName/produce/single", handler.CreateHandleMessage(conn))
12+
api := app.Group("/stations", logger.New())
13+
api.Post("/:stationName/produce/single", handler.CreateHandleMessage(conn))
1414
}

0 commit comments

Comments
 (0)