feat(bulkdata): new folder organization

backend/services/acs/README.md

@@ -1,97 +0,0 @@
-# Moses ACS [![Build Status](https://travis-ci.org/lucacervasio/mosesacs.svg?branch=master)](https://travis-ci.org/lucacervasio/mosesacs)
-
-An ACS in Go for provisioning CPEs, suitable for test purposes or production deployment.
-
-## Getting started
-
-Install the package:
-
-    go get oktopUSP/backend/services/acs
-
-Run daemon:
-
-    mosesacs -d
-
-Connect to it and get a cli:
-
-    mosesacs
-
-Congratulations, you've connected to the daemon via websocket. Now you can issue commands via CLI or browse the embedded webserver at http://localhost:9292/www
-
-## Compatibility on ARM
-
-Moses is built on purpose only with dependencies in pure GO. So it runs on ARM processors with no issues. We tested it on QNAP devices and Raspberry for remote control.
-
-## CLI commands
-
-### 1. `list`: list CPEs
-
- example:
-
-```
- moses@localhost:9292/> list
- cpe list
- CPE A54FD with OUI 006754
-```
-
-### 2. `readMib SERIAL LEAF/SUBTREE`: read a specific leaf or a subtree
-
- example:
-
-```
- moses@localhost:9292/> readMib A54FD Device.
- Received an Inform from [::1]:58582 (3191 bytes) with SerialNumber A54FD and EventCodes 6 CONNECTION REQUEST
- InternetGatewayDevice.Time.NTPServer1 : pool.ntp.org
- InternetGatewayDevice.Time.CurrentLocalTime : 2014-07-11T09:08:25
- InternetGatewayDevice.Time.LocalTimeZone : +00:00
- InternetGatewayDevice.Time.LocalTimeZoneName : Greenwich Mean Time : Dublin
- InternetGatewayDevice.Time.DaylightSavingsUsed : 0
-```
-
-### 3. `writeMib SERIAL LEAF VALUE`: issue a SetParameterValues and write a value into a leaf
-
- example:
-
-```
- moses@localhost:9292/> writeMib A54FD InternetGatewayDevice.Time.Enable false
- Received an Inform from [::1]:58582 (3191 bytes) with SerialNumber A54FD and EventCodes 6 CONNECTION REQUEST
-```
-
-### 4. `GetParameterNames SERIAL LEAF/SUBTREE`: issue a GetParameterNames and get all leaves/objects at first level
-
- example:
-
-```
-moses@localhost:9292/> GetParameterNames A54FD InternetGatewayDevice.
-Received an Inform from [::1]:55385 (3119 bytes) with SerialNumber A54FD and EventCodes 6 CONNECTION REQUEST
-InternetGatewayDevice.LANDeviceNumberOfEntries : 0
-InternetGatewayDevice.WANDeviceNumberOfEntries : 0
-InternetGatewayDevice.DeviceInfo. : 0
-InternetGatewayDevice.ManagementServer. : 0
-InternetGatewayDevice.Time. : 0
-InternetGatewayDevice.Layer3Forwarding. : 0
-InternetGatewayDevice.LANDevice. : 0
-InternetGatewayDevice.WANDevice. : 0
-InternetGatewayDevice.X_00507F_InternetAcc. : 0
-InternetGatewayDevice.X_00507F_LAN. : 0
-InternetGatewayDevice.X_00507F_NAT. : 0
-InternetGatewayDevice.X_00507F_VLAN. : 0
-InternetGatewayDevice.X_00507F_Firewall. : 0
-InternetGatewayDevice.X_00507F_Applications. : 0
-InternetGatewayDevice.X_00507F_System. : 0
-InternetGatewayDevice.X_00507F_Status. : 0
-InternetGatewayDevice.X_00507F_Diagnostics. : 0
-```
-
-
-
-
-## Services exposed
-
-Moses exposes three services:
-
- - http://localhost:9292/acs is the endpoint for the CPEs to connect
- - http://localhost:9292/www is the embedded webserver to control your CPEs
- - ws://localhost:9292/ws is the websocket endpoint used by the cli to issue commands. Read about the API specification if you want to build a custom frontend which interacts with mosesacs daemon.
-
-

backend/services/bulkdata/http/build/Dockerfile

@@ -1,8 +1,8 @@
 FROM golang:1.22@sha256:82e07063a1ac3ee59e6f38b1222e32ce88469e4431ff6496cc40fb9a0fc18229 as builder
 WORKDIR /app
 COPY ../ .
-RUN CGO_ENABLED=0 GOOS=linux go build -o bulkdata cmd/bulkdata/main.go
+RUN CGO_ENABLED=0 GOOS=linux go build -o http-bulk-collector cmd/http-bulk-collector/main.go
 
 FROM alpine:3.14@sha256:0f2d5c38dd7a4f4f733e688e3a6733cb5ab1ac6e3cb4603a5dd564e5bfb80eed
-COPY --from=builder /app/bulkdata /
+COPY --from=builder /app/http-bulk-collector /
 ENTRYPOINT ["/bulkdata"]

backend/services/bulkdata/http/cmd/http-bulk-collector/main.go

@@ -7,6 +7,7 @@ import (
 	"syscall"
 
 	"github.com/oktopUSP/backend/services/bulkdata/internal/api"
+	"github.com/oktopUSP/backend/services/bulkdata/internal/bridge"
 	"github.com/oktopUSP/backend/services/bulkdata/internal/config"
 	"github.com/oktopUSP/backend/services/bulkdata/internal/nats"
 )
@@ -18,9 +19,11 @@ func main() {
 
 	c := config.NewConfig()
 
-	js, nc, kv := nats.StartNatsClient(c.Nats)
+	pub, sub := nats.StartNatsClient(c.Nats)
 
-	server := api.NewApi(c.RestApi, js, nc, kv)
+	b := bridge.NewBridge(pub, sub)
+
+	server := api.NewApi(c.RestApi, b)
 
 	server.StartApi()
 

backend/services/bulkdata/http/internal/api/api.go

@@ -6,11 +6,10 @@ import (
 	"time"
 
 	"github.com/gorilla/mux"
-	"github.com/nats-io/nats.go"
-	"github.com/nats-io/nats.go/jetstream"
 	"github.com/oktopUSP/backend/services/bulkdata/internal/api/cors"
 	"github.com/oktopUSP/backend/services/bulkdata/internal/api/handler"
 	"github.com/oktopUSP/backend/services/bulkdata/internal/api/middleware"
+	"github.com/oktopUSP/backend/services/bulkdata/internal/bridge"
 	"github.com/oktopUSP/backend/services/bulkdata/internal/config"
 )
 
@@ -21,10 +20,10 @@ type Api struct {
 
 const REQUEST_TIMEOUT = time.Second * 30
 
-func NewApi(c config.RestApi, js jetstream.JetStream, nc *nats.Conn, kv jetstream.KeyValue) Api {
+func NewApi(c config.RestApi, b *bridge.Bridge) Api {
 	return Api{
 		port:    c.Port,
-		handler: handler.NewHandler(js, nc, kv),
+		handler: handler.NewHandler(c.Ctx, b),
 	}
 }
 

backend/services/bulkdata/http/internal/api/handler/handler.go

@@ -0,0 +1,19 @@
+package handler
+
+import (
+	"context"
+
+	"github.com/oktopUSP/backend/services/bulkdata/internal/bridge"
+)
+
+type Handler struct {
+	ctx context.Context
+	b   *bridge.Bridge
+}
+
+func NewHandler(ctx context.Context, b *bridge.Bridge) Handler {
+	return Handler{
+		ctx: ctx,
+		b:   b,
+	}
+}

backend/services/bulkdata/http/internal/bridge/bridge.go

@@ -0,0 +1,28 @@
+package bridge
+
+import (
+	"github.com/nats-io/nats.go"
+)
+
+type (
+	Publisher  func(string, []byte) error
+	Subscriber func(string, func(*nats.Msg)) error
+)
+
+type Bridge struct {
+	pub Publisher
+	sub Subscriber
+}
+
+const BULK_DATA_SUBJECT = "bulk"
+
+func NewBridge(p Publisher, s Subscriber) *Bridge {
+	return &Bridge{
+		pub: p,
+		sub: s,
+	}
+}
+
+func (b *Bridge) SendDeviceData(deviceId string, payload []byte) error {
+	return b.pub("oi", payload)
+}

backend/services/bulkdata/http/internal/nats/nats.go

@@ -5,16 +5,13 @@ import (
 	"time"
 
 	"github.com/nats-io/nats.go"
-	"github.com/nats-io/nats.go/jetstream"
 	"github.com/oktopUSP/backend/services/bulkdata/internal/config"
 )
 
-const (
-	BUCKET_NAME        = "devices-auth"
-	BUCKET_DESCRIPTION = "Devices authentication"
-)
-
-func StartNatsClient(c config.Nats) (jetstream.JetStream, *nats.Conn, jetstream.KeyValue) {
+func StartNatsClient(c config.Nats) (
+	func(string, []byte) error,
+	func(string, func(*nats.Msg)) error,
+) {
 
 	var (
 		nc  *nats.Conn
@@ -35,20 +32,27 @@ func StartNatsClient(c config.Nats) (jetstream.JetStream, *nats.Conn, jetstream.
 	}
 	log.Printf("Successfully connected to NATS server %s", c.Url)
 
-	js, err := jetstream.New(nc)
-	if err != nil {
-		log.Fatalf("Failed to create JetStream client: %v", err)
-	}
+	return publisher(nc), subscriber(nc)
+}
 
-	kv, err := js.CreateOrUpdateKeyValue(c.Ctx, jetstream.KeyValueConfig{
-		Bucket:      BUCKET_NAME,
-		Description: BUCKET_DESCRIPTION,
-	})
-	if err != nil {
-		log.Fatalf("Failed to create KeyValue store: %v", err)
+func subscriber(nc *nats.Conn) func(string, func(*nats.Msg)) error {
+	return func(subject string, handler func(*nats.Msg)) error {
+		_, err := nc.Subscribe(subject, handler)
+		if err != nil {
+			log.Printf("error to subscribe to subject %s error: %q", subject, err)
+		}
+		return err
 	}
+}
 
-	return js, nc, kv
+func publisher(nc *nats.Conn) func(string, []byte) error {
+	return func(subject string, payload []byte) error {
+		err := nc.Publish(subject, payload)
+		if err != nil {
+			log.Printf("error to send nats core message: %q", err)
+		}
+		return err
+	}
 }
 
 func defineOptions(c config.Nats) []nats.Option {

backend/services/bulkdata/internal/api/handler/handler.go

@@ -1,23 +0,0 @@
-package handler
-
-import (
-	"context"
-
-	"github.com/nats-io/nats.go"
-	"github.com/nats-io/nats.go/jetstream"
-)
-
-type Handler struct {
-	js  jetstream.JetStream
-	nc  *nats.Conn
-	kv  jetstream.KeyValue
-	ctx context.Context
-}
-
-func NewHandler(js jetstream.JetStream, nc *nats.Conn, kv jetstream.KeyValue) Handler {
-	return Handler{
-		js: js,
-		nc: nc,
-		kv: kv,
-	}
-}