Fix build issue with server

Not useful yet... I thought it would be but I changed my mind...
Need to think about this a lot more

.gitignore

@@ -1,3 +1,4 @@
 jstester/node_modules
 encoding/tmp
 pdu/tmp
+tmp

client/client.go

@@ -1 +1,165 @@
 package client
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"net"
+	"smpptester/pdu"
+	"smpptester/utils"
+	"time"
+
+	"golang.org/x/time/rate"
+)
+
+var (
+	SMPP_CLIENT_RETRY_TIMEOUT      = 1 * time.Second
+	SMPP_CLIENT_SEND_QUEUE_SIZE    = 1024 * 8
+	SMPP_CLIENT_RECEIVE_QUEUE_SIZE = 1024 * 8
+	SMPP_CLIENT_ASYNC_SEND_WORKERS = 16
+	SMPP_CLIENT_SEND_LIMIT         = 4
+)
+
+type SMPPClient struct {
+	Id           int
+	Connected    bool
+	SendQueue    chan pdu.PDU
+	ReceiveQueue chan pdu.PDU
+	SendLimit    *utils.ReactiveValue[int]
+
+	sendLimiter *rate.Limiter
+	enabled     bool
+	running     bool
+	conn        net.Conn
+	port        string
+	log         log.Logger
+	dcErr       chan (error)
+}
+
+func NewSMPPClient(port string, id int) *SMPPClient {
+	client := &SMPPClient{
+		Id:           id,
+		SendQueue:    make(chan pdu.PDU, SMPP_CLIENT_SEND_QUEUE_SIZE),
+		ReceiveQueue: make(chan pdu.PDU, SMPP_CLIENT_RECEIVE_QUEUE_SIZE),
+		SendLimit:    utils.NewReactiveValue[int](SMPP_CLIENT_SEND_LIMIT),
+
+		sendLimiter: rate.NewLimiter(rate.Limit(SMPP_CLIENT_SEND_LIMIT), SMPP_CLIENT_SEND_LIMIT),
+		port:        port,
+		log:         log.Logger{},
+		dcErr:       make(chan error, 128),
+	}
+	client.log = *log.New(log.Writer(), "", log.LstdFlags|log.Lshortfile)
+	client.log.SetPrefix(fmt.Sprintf("SMPP client %d: ", client.Id))
+
+	go func() {
+		limit := client.SendLimit.Subscribe()
+		defer client.SendLimit.Unsubscribe(limit)
+		for {
+			v := <-limit
+			client.log.Printf("Send limit changed to %d", *v)
+			client.sendLimiter.SetLimit(rate.Limit(*v))
+			client.sendLimiter.SetBurst(*v)
+		}
+	}()
+
+	return client
+}
+
+func (c *SMPPClient) Enable() {
+	if c.enabled {
+		c.log.Printf("SMPP client %d is already enabled", c.Id)
+		return
+	}
+	c.enabled = true
+	c.start()
+	c.log.Printf("SMPP client %d enabled", c.Id)
+}
+func (c *SMPPClient) Disable() {
+	if !c.enabled {
+		c.log.Printf("SMPP client %d is already disabled", c.Id)
+		return
+	}
+	c.enabled = false
+	c.dcErr <- fmt.Errorf("client is disabled")
+	if c.conn != nil {
+		c.conn.Close()
+	}
+	c.log.Printf("SMPP client %d disabled", c.Id)
+}
+
+func (c *SMPPClient) start() {
+	if c.running {
+		c.log.Printf("SMPP client %d is already running", c.Id)
+		return
+	}
+	go func() {
+		c.running = true
+		defer func() {
+			c.running = false
+		}()
+
+		c.log.Printf("SMPP client %d started", c.Id)
+		for {
+			if c.enabled && !c.Connected {
+				c.log.Printf("Trying to connect to %s", c.port)
+				err := c.connect()
+				if err != nil {
+					c.dcErr <- err
+				}
+			}
+			err := <-c.dcErr
+			c.Connected = false
+			if c.enabled {
+				c.log.Printf("Disconnected: '%v' trying again in %d second(s)", err, int(SMPP_CLIENT_RETRY_TIMEOUT.Seconds()))
+				time.Sleep(SMPP_CLIENT_RETRY_TIMEOUT)
+			} else {
+				c.log.Printf("Disconnected: '%v' & client is disabled, quitting", err)
+				break
+			}
+		}
+	}()
+}
+
+func (c *SMPPClient) connect() error {
+	if !c.enabled {
+		return fmt.Errorf("client is not enabled")
+	}
+	if c.Connected {
+		return fmt.Errorf("client is already connected")
+	}
+	conn, err := net.Dial("tcp", c.port)
+	if err != nil {
+		return fmt.Errorf("failed to connect to SMPP server: %w", err)
+	}
+	c.log.Printf("SMPP client %d connected to %s", c.Id, c.port)
+	c.Connected = true
+	c.conn = conn
+	return nil
+}
+
+func (c *SMPPClient) Send(pdata pdu.PDU) error {
+	if c.conn == nil {
+		return fmt.Errorf("connection is not established")
+	}
+
+	pdata.UpdateSize()
+	buf := pdu.ByteBufferPool.Get(pdata.Size())
+	err := pdata.Encode(buf)
+	if err != nil {
+		return fmt.Errorf("failed to encode PDU: %w", err)
+	}
+
+	err = c.sendLimiter.Wait(context.Background())
+	if err != nil {
+		return fmt.Errorf("failed to wait for rate limiter: %w", err)
+	}
+
+	_, err = c.conn.Write(buf.Bytes())
+	if err != nil {
+		c.dcErr <- err
+		return fmt.Errorf("failed to send PDU: %w", err)
+	}
+
+	// c.log.Printf("SMPP client %d sent PDU: %+v", c.Id, pdata)
+	return nil
+}

go.mod

@@ -2,4 +2,7 @@ module smpptester
 
 go 1.22.4
 
-require github.com/yuin/gopher-lua v1.1.1
+require (
+	github.com/yuin/gopher-lua v1.1.1
+	golang.org/x/time v0.5.0
+)

go.sum

@@ -1,2 +1,4 @@
 github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M=
 github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw=
+golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
+golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=

main.go

@@ -4,9 +4,10 @@ import (
 	"bytes"
 	"encoding/binary"
 	"log"
-	"net"
 	"sync"
+	"time"
 
+	"smpptester/client"
 	"smpptester/pdu"
 )
 
@@ -166,47 +167,26 @@ func main() {
 	buf := pdu.ByteBufferPool.Get(submit.Size())
 	defer pdu.ByteBufferPool.Put(buf)
 	submit.Encode(buf)
-	log.Println(buf.Bytes())
-	log.Println(createSubmitSMPDU())
-	// log.Println(pdu.Encode())
-	// serviceType := "CMT"
-	// sourceAddrTON := byte(0x01)
-	// sourceAddrNPI := byte(0x01)
-	// sourceAddr := "12345"
-	// destAddrTON := byte(0x01)
-	// destAddrNPI := byte(0x01)
-	// destinationAddr := "67890"
-	// esmClass := byte(0x00)
-	// protocolID := byte(0x00)
-	// priorityFlag := byte(0x00)
-	// scheduleDeliveryTime := ""
-	// validityPeriod := ""
-	// registeredDelivery := byte(0x01)
-	// replaceIfPresentFlag := byte(0x00)
-	// dataCoding := byte(0x01)
-	// smDefaultMsgID := byte(0x00)
-	// shortMessage := "Hello, SMPP!"
 
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
 
-	conn, err := net.Dial("tcp", "localhost:2775")
-	if err != nil {
-		log.Printf("Failed to connect to SMPP server: %+v", err)
-		return
-	}
-	log.Printf("Connected to SMPP server")
-	defer conn.Close()
-
-	go func() {
-		conn.Write(buf.Bytes())
-		data, err := conn.Read(make([]byte, 1024))
-		if err != nil {
-			log.Printf("Failed to read from connection: %+v", err)
-			return
+	client := client.NewSMPPClient("localhost:2775", 1)
+	client.Enable()	
+	for {
+		if client.Connected {
+			// err := client.Send(submit)
+			// if err != nil {
+			// 	log.Printf("Failed to send PDU: %v", err)
+			// }
+			go client.Send(submit)
+			// if rand.Int() % 1000 == 0 {
+			// 	client.SendLimit.Set(rand.Int() % 1000)
+			// }
 		}
-		log.Printf("Received data: %#v", data)
-	}()
+		time.Sleep(500 * time.Millisecond)
+	}
+	log.Println("Are we done")
 
 	wg.Wait()
 }

server/server.go

@@ -1 +1,66 @@
 package server
+
+import (
+	"fmt"
+	"log"
+	"net"
+)
+
+type SMPPServer struct {
+	Id int
+	conns []*net.Conn
+	listener net.Listener
+	port  string
+	log log.Logger
+}
+
+func NewSMPPServer(port string, id int) *SMPPServer {
+	server := &SMPPServer{
+		port: port,
+		log: log.Logger{},
+		Id: id,
+	}
+	server.log = *log.New(log.Writer(), "", log.LstdFlags)
+	server.log.SetPrefix(fmt.Sprintf("SMPP server %d: ", server.Id))
+	return server
+}
+
+func (s *SMPPServer) Listen() error {
+	listener, err := net.Listen("tcp", s.port)
+	if err != nil {
+		return fmt.Errorf("failed to start SMPP server: %w", err)
+	}
+	// s.log.Printf("SMPP server %d started on %s", s.port)
+	s.listener = listener
+	return nil
+}
+
+// package main
+
+// import (
+// 	"net"
+// 	"fmt"
+// )
+
+// func main() {
+// 	ln, err := net.Listen("tcp", ":8080")
+// 	if err != nil {
+// 		fmt.Println(err)
+// 		return
+// 	}
+// 	defer ln.Close()
+
+// 	for {
+// 		conn, err := ln.Accept()
+// 		if err != nil {
+// 			fmt.Println(err)
+// 			continue
+// 		}
+// 		go handleConnection(conn)
+// 	}
+// }
+
+// func handleConnection(conn net.Conn) {
+// 	defer conn.Close()
+// 	// Handle the connection
+// }

utils/reactivevalue.go

@@ -0,0 +1,36 @@
+package utils
+
+type ReactiveValue[T any] struct {
+	V       T
+	subscribers []chan *T
+}
+
+func NewReactiveValue[T any](value T) *ReactiveValue[T] {
+	return &ReactiveValue[T]{
+		V:       value,
+		subscribers: make([]chan *T, 0),
+	}
+}
+
+func (rv *ReactiveValue[T]) Set(value T) {
+	rv.V = value
+	for _, subscriber := range rv.subscribers {
+		subscriber <- &value
+	}
+}
+
+func (rv *ReactiveValue[T]) Subscribe() chan *T {
+	subscriber := make(chan *T)
+	rv.subscribers = append(rv.subscribers, subscriber)
+	return subscriber
+}
+
+func (rv *ReactiveValue[T]) Unsubscribe(subscriber chan *T) {
+	for i, s := range rv.subscribers {
+		if s == subscriber {
+			rv.subscribers = append(rv.subscribers[:i], rv.subscribers[i+1:]...)
+			close(subscriber)
+			return
+		}
+	}
+}