Merge pull request #11 from gatewayd-io/add-ci

Add CI with lint, test and (commented out) coverage reporting

Author: mostafa

.github/workflows/test.yaml

@@ -0,0 +1,36 @@
+name: Test GatewayD
+
+on:
+  push:
+    branches:
+      - main
+    tags:
+      - v*
+  pull_request:
+
+jobs:
+  test:
+    name: Test GatewayD
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout 🛎️
+        uses: actions/checkout@v3
+        with:
+          fetch-depth: 0
+
+      - name: Install Go 🧑‍💻
+        uses: actions/setup-go@v3
+        with:
+          go-version: '1.18'
+
+      - name: Lint code issues 🚨
+        uses: golangci/golangci-lint-action@v3
+
+      - name: Run Go tests 🔬
+        run: go test -cover -covermode atomic -coverprofile=profile.cov -v ./...
+
+      # Enable coverage reporting
+      # - name: Report coverage to coveralls 📈
+      #   uses: shogo82148/actions-goveralls@v1
+      #   with:
+      #     path-to-profile: profile.cov

.golangci.yaml

@@ -0,0 +1,17 @@
+linters:
+  enable-all: true
+  disable:
+    - cyclop
+    - wsl
+    - godox
+    - gochecknoglobals
+    - ireturn
+    - nlreturn
+    - testpackage
+    - paralleltest
+    - exhaustivestruct
+    - exhaustruct
+    - gocognit
+    - gochecknoinits
+    - gocyclo
+

main.go

@@ -8,15 +8,27 @@ import (
 
 	"github.com/gatewayd-io/gatewayd/network"
 	"github.com/panjf2000/gnet/v2"
+	"github.com/sirupsen/logrus"
 )
 
+const (
+	DefaultTCPKeepAlive = 3 * time.Second
+)
+
+//nolint:funlen
 func main() {
 	// Create a server
-	server := &network.Server{
-		Network: "tcp",
-		Address: "0.0.0.0:15432",
-		Status:  network.Stopped,
-		Options: []gnet.Option{
+	server := network.NewServer(
+		"tcp",
+		"0.0.0.0:15432",
+		0,
+		0,
+		network.DefaultTickInterval,
+		network.DefaultPoolSize,
+		network.DefaultBufferSize,
+		false,
+		false,
+		[]gnet.Option{
 			// Scheduling options
 			gnet.WithMulticore(true),
 			gnet.WithLockOSThread(false),
@@ -37,18 +49,20 @@ func main() {
 
 			// Buffer options
 			// TODO: This should be configurable and optimized.
-			gnet.WithReadBufferCap(4096),
-			gnet.WithWriteBufferCap(4096),
-			gnet.WithSocketRecvBuffer(4096),
-			gnet.WithSocketSendBuffer(4096),
+			gnet.WithReadBufferCap(network.DefaultBufferSize),
+			gnet.WithWriteBufferCap(network.DefaultBufferSize),
+			gnet.WithSocketRecvBuffer(network.DefaultBufferSize),
+			gnet.WithSocketSendBuffer(network.DefaultBufferSize),
 
 			// TCP options
 			gnet.WithReuseAddr(true),
 			gnet.WithReusePort(true),
-			gnet.WithTCPKeepAlive(time.Second * 3),
+			gnet.WithTCPKeepAlive(DefaultTCPKeepAlive),
 			gnet.WithTCPNoDelay(gnet.TCPNoDelay),
 		},
-	}
+		nil,
+		nil,
+	)
 
 	// Shutdown the server gracefully
 	var signals []os.Signal
@@ -75,5 +89,7 @@ func main() {
 	}()
 
 	// Run the server
-	server.Run()
+	if err := server.Run(); err != nil {
+		logrus.Error(err)
+	}
 }

network/client.go

@@ -1,19 +1,23 @@
 package network
 
 import (
+	"fmt"
 	"net"
 
 	"github.com/sirupsen/logrus"
 )
 
+const (
+	DefaultSeed = 1000
+)
+
 type Client struct {
 	net.Conn
 
 	ID                string
 	ReceiveBufferSize int
 	Network           string // tcp/udp/unix
 	Address           string
-
 	// TODO: add read/write deadline and deal with timeouts
 }
 
@@ -51,19 +55,18 @@ func NewClient(network, address string, receiveBufferSize int) *Client {
 
 	client.Conn = conn
 	if client.ReceiveBufferSize == 0 {
-		client.ReceiveBufferSize = 4096
+		client.ReceiveBufferSize = DefaultBufferSize
 	}
 	logrus.Debugf("New client created: %s", client.Address)
-	client.ID = GetID(conn.LocalAddr().Network(), conn.LocalAddr().String(), 1000)
+	client.ID = GetID(conn.LocalAddr().Network(), conn.LocalAddr().String(), DefaultSeed)
 
 	return &client
 }
 
 func (c *Client) Send(data []byte) error {
-	_, err := c.Write(data)
-	if err != nil {
+	if _, err := c.Write(data); err != nil {
 		logrus.Errorf("Couldn't send data to the server: %s", err)
-		return err
+		return fmt.Errorf("couldn't send data to the server: %w", err)
 	}
 	logrus.Debugf("Sent %d bytes to %s", len(data), c.Address)
 	// logrus.Infof("Sent data: %s", data)
@@ -75,7 +78,7 @@ func (c *Client) Receive() (int, []byte, error) {
 	read, err := c.Read(buf)
 	if err != nil {
 		logrus.Errorf("Couldn't receive data from the server: %s", err)
-		return 0, nil, err
+		return 0, nil, fmt.Errorf("couldn't receive data from the server: %w", err)
 	}
 	logrus.Debugf("Received %d bytes from %s", read, c.Address)
 	// logrus.Infof("Received data: %s", buf[:read])

network/client_test.go

@@ -19,14 +19,14 @@ func TestNewClient(t *testing.T) {
 		}
 	}()
 
-	c := NewClient("tcp", "localhost:5432", 4096)
-	defer c.Close()
-
-	assert.Equal(t, "tcp", c.Network)
-	assert.Equal(t, "127.0.0.1:5432", c.Address)
-	assert.Equal(t, 4096, c.ReceiveBufferSize)
-	assert.NotEmpty(t, c.ID)
-	assert.NotNil(t, c.Conn)
+	client := NewClient("tcp", "localhost:5432", DefaultBufferSize)
+	defer client.Close()
+
+	assert.Equal(t, "tcp", client.Network)
+	assert.Equal(t, "127.0.0.1:5432", client.Address)
+	assert.Equal(t, DefaultBufferSize, client.ReceiveBufferSize)
+	assert.NotEmpty(t, client.ID)
+	assert.NotNil(t, client.Conn)
 }
 
 func TestSend(t *testing.T) {
@@ -41,11 +41,11 @@ func TestSend(t *testing.T) {
 		}
 	}()
 
-	c := NewClient("tcp", "localhost:5432", 4096)
-	defer c.Close()
+	client := NewClient("tcp", "localhost:5432", DefaultBufferSize)
+	defer client.Close()
 
-	assert.NotNil(t, c)
-	err := c.Send(CreatePostgreSQLPacket('Q', []byte("select 1;")))
+	assert.NotNil(t, client)
+	err := client.Send(CreatePostgreSQLPacket('Q', []byte("select 1;")))
 	assert.Nil(t, err)
 }
 
@@ -61,22 +61,22 @@ func TestReceive(t *testing.T) {
 		}
 	}()
 
-	c := NewClient("tcp", "localhost:5432", 4096)
-	defer c.Close()
+	client := NewClient("tcp", "localhost:5432", DefaultBufferSize)
+	defer client.Close()
 
-	assert.NotNil(t, c)
-	err := c.Send(CreatePostgreSQLPacket('Q', []byte("select 1;")))
+	assert.NotNil(t, client)
+	err := client.Send(CreatePgStartupPacket())
 	assert.Nil(t, err)
 
-	size, data, err := c.Receive()
-	msg := "SFATAL\x00VFATAL\x00C0A000\x00Munsupported frontend protocol 0.0: server supports 3.0 to 3.0\x00Fpostmaster.c\x00L2138\x00RProcessStartupPacket\x00\x00"
-	assert.Equal(t, 132, size)
+	size, data, err := client.Receive()
+	msg := "\x00\x00\x00\x03"
+	assert.Equal(t, 9, size)
 	assert.Equal(t, len(data[:size]), size)
 	assert.Nil(t, err)
 	assert.NotEmpty(t, data[:size])
 	assert.Equal(t, msg, string(data[5:size]))
-	assert.Equal(t, "E", string(data[0]))
-	assert.Equal(t, 83, int(data[5]))
+	// AuthenticationOk
+	assert.Equal(t, uint8(0x52), data[0])
 }
 
 func TestClose(t *testing.T) {
@@ -91,12 +91,12 @@ func TestClose(t *testing.T) {
 		}
 	}()
 
-	c := NewClient("tcp", "localhost:5432", 4096)
-	assert.NotNil(t, c)
-	c.Close()
-	assert.Equal(t, "", c.ID)
-	assert.Equal(t, "", c.Network)
-	assert.Equal(t, "", c.Address)
-	assert.Nil(t, c.Conn)
-	assert.Equal(t, 0, c.ReceiveBufferSize)
+	client := NewClient("tcp", "localhost:5432", DefaultBufferSize)
+	assert.NotNil(t, client)
+	client.Close()
+	assert.Equal(t, "", client.ID)
+	assert.Equal(t, "", client.Network)
+	assert.Equal(t, "", client.Address)
+	assert.Nil(t, client.Conn)
+	assert.Equal(t, 0, client.ReceiveBufferSize)
 }

network/errors.go

@@ -0,0 +1,10 @@
+package network
+
+import "errors"
+
+var (
+	ErrClientNotFound      = errors.New("client not found")
+	ErrNetworkNotSupported = errors.New("network is not supported")
+	ErrClientNotConnected  = errors.New("client is not connected")
+	ErrPoolExhausted       = errors.New("pool is exhausted")
+)

network/network_helpers_test.go

@@ -1,13 +1,56 @@
 package network
 
-import "encoding/binary"
+import (
+	"encoding/binary"
+)
+
+type WriteBuffer struct {
+	Bytes []byte
+
+	msgStart int
+}
+
+func writeStartupMsg(buf *WriteBuffer, user, database, appName string) {
+	// Write startup message header
+	buf.msgStart = len(buf.Bytes)
+	buf.Bytes = append(buf.Bytes, 0, 0, 0, 0)
+
+	// Write protocol version
+	buf.Bytes = append(buf.Bytes, 0, 0, 0, 0)
+	binary.BigEndian.PutUint32(buf.Bytes[len(buf.Bytes)-4:], uint32(196608))
+
+	buf.WriteString("user")
+	buf.WriteString(user)
+	buf.WriteString("database")
+	buf.WriteString(database)
+	buf.WriteString("application_name")
+	buf.WriteString(appName)
+	buf.WriteString("")
+
+	// Write message length
+	binary.BigEndian.PutUint32(
+		buf.Bytes[buf.msgStart:], uint32(len(buf.Bytes)-buf.msgStart))
+}
+
+func (buf *WriteBuffer) WriteString(s string) {
+	buf.Bytes = append(buf.Bytes, s...)
+	buf.Bytes = append(buf.Bytes, 0)
+}
 
 func CreatePostgreSQLPacket(typ byte, msg []byte) []byte {
 	packet := make([]byte, 1+4+len(msg))
 
-	packet = append(packet, typ)
-	binary.BigEndian.PutUint32(packet, uint32(len(msg)+4))
-	packet = append(packet, msg...)
+	packet[0] = typ
+	binary.BigEndian.PutUint32(packet[1:], uint32(len(msg)+4))
+	for i, b := range msg {
+		packet[i+5] = b
+	}
 
 	return packet
 }
+
+func CreatePgStartupPacket() []byte {
+	buf := &WriteBuffer{}
+	writeStartupMsg(buf, "postgres", "postgres", "gatewayd")
+	return buf.Bytes
+}