[api_gateway]Add connection query API

api_gateway/connection_query_api.go

@@ -0,0 +1,124 @@
+package api_gateway
+
+import (
+	"errors"
+	"sync"
+
+	"github.com/it-chain/engine/common/event"
+)
+
+var ErrConnectionExists = errors.New("Connection already exists")
+
+type ConnectionQueryApi struct {
+	mux                  *sync.Mutex
+	connectionRepository ConnectionRepository
+}
+
+func NewConnectionQueryApi(connRepo ConnectionRepository) *ConnectionQueryApi {
+	return &ConnectionQueryApi{
+		mux:                  &sync.Mutex{},
+		connectionRepository: connRepo,
+	}
+}
+
+func (q ConnectionQueryApi) GetAllConnectionList() ([]Connection, error) {
+	return q.connectionRepository.FindAll()
+}
+
+func (q ConnectionQueryApi) GetConnectionByID(connID string) (Connection, error) {
+	return q.connectionRepository.FindByID(connID)
+}
+
+type ConnectionRepository struct {
+	mux             *sync.RWMutex
+	ConnectionTable map[string]Connection
+}
+
+func NewConnectionRepository() *ConnectionRepository {
+	return &ConnectionRepository{
+		mux:             &sync.RWMutex{},
+		ConnectionTable: make(map[string]Connection),
+	}
+}
+
+func (cr *ConnectionRepository) Save(conn Connection) error {
+	cr.mux.Lock()
+	defer cr.mux.Unlock()
+
+	_, exist := cr.ConnectionTable[conn.ConnectionID]
+	if exist {
+		return ErrConnectionExists
+	}
+
+	cr.ConnectionTable[conn.ConnectionID] = conn
+
+	return nil
+}
+
+func (cr *ConnectionRepository) Remove(connID string) error {
+	cr.mux.Lock()
+	defer cr.mux.Unlock()
+
+	delete(cr.ConnectionTable, connID)
+
+	return nil
+}
+
+func (cr *ConnectionRepository) FindAll() ([]Connection, error) {
+	cr.mux.Lock()
+	defer cr.mux.Unlock()
+
+	connectionList := []Connection{}
+
+	for _, conn := range cr.ConnectionTable {
+		connectionList = append(connectionList, conn)
+	}
+
+	return connectionList, nil
+}
+
+func (cr *ConnectionRepository) FindByID(connID string) (Connection, error) {
+	cr.mux.Lock()
+	defer cr.mux.Unlock()
+
+	for _, conn := range cr.ConnectionTable {
+		if connID == conn.ConnectionID {
+			return conn, nil
+		}
+	}
+
+	return Connection{}, nil
+}
+
+type ConnectionEventListener struct {
+	connectionRepository ConnectionRepository
+}
+
+func NewConnectionEventListener(connRepo ConnectionRepository) ConnectionEventListener {
+	return ConnectionEventListener{
+		connectionRepository: connRepo,
+	}
+}
+
+func (cel *ConnectionEventListener) HandleConnectionCreatedEvent(event event.ConnectionCreated) error {
+	connection := Connection{
+		ConnectionID: event.ConnectionID,
+		Address:      event.Address,
+	}
+
+	err := cel.connectionRepository.Save(connection)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (cel *ConnectionEventListener) HandleConnectionClosedEvent(event event.ConnectionClosed) error {
+	cel.connectionRepository.Remove(event.ConnectionID)
+	return nil
+}
+
+type Connection struct {
+	ConnectionID string
+	Address      string
+}

api_gateway/connection_query_api_test.go

@@ -0,0 +1,163 @@
+package api_gateway_test
+
+import (
+	"strconv"
+	"testing"
+
+	"github.com/it-chain/engine/api_gateway"
+	"github.com/it-chain/engine/common/event"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestConnectionQueryApi_GetAllConnectionList(t *testing.T) {
+	repo := api_gateway.NewConnectionRepository()
+	api := api_gateway.NewConnectionQueryApi(*repo)
+
+	for i := 0; i < 10; i++ {
+		id := strconv.Itoa(i)
+		err := repo.Save(api_gateway.Connection{
+			ConnectionID: id,
+			Address:      "address" + id,
+		})
+		assert.NoError(t, err)
+	}
+
+	connectionList, err := api.GetAllConnectionList()
+	assert.NoError(t, err)
+
+	for i := 0; i < 11; i++ {
+		var id string
+		if i != 10 {
+			id = connectionList[i].ConnectionID
+		} else {
+			id = strconv.Itoa(i)
+		}
+
+		connection, err := api.GetConnectionByID(id)
+		assert.NoError(t, err)
+
+		if i != 10 {
+			assert.NotEqual(t, api_gateway.Connection{}, connection)
+		} else {
+			assert.Equal(t, api_gateway.Connection{}, connection)
+		}
+	}
+}
+
+func TestConnectionQueryApi_GetConnectionByID(t *testing.T) {
+	repo := api_gateway.NewConnectionRepository()
+	api := api_gateway.NewConnectionQueryApi(*repo)
+	ids := [4]string{"it-chain", "engine", "api_gateway", "makehoney"}
+
+	for i := 0; i < len(ids); i++ {
+		err := repo.Save(api_gateway.Connection{
+			ConnectionID: ids[i],
+			Address:      "address",
+		})
+		assert.NoError(t, err)
+	}
+
+	for i := 0; i < len(ids); i++ {
+		connection, err := api.GetConnectionByID(ids[i])
+		assert.NoError(t, err)
+
+		assert.NotEqual(t, api_gateway.Connection{}, connection)
+	}
+	connection, err := api.GetConnectionByID("wrongID")
+	assert.NoError(t, err)
+	assert.Equal(t, api_gateway.Connection{}, connection)
+}
+
+func TestConnectionRepository_Save(t *testing.T) {
+	repo := api_gateway.NewConnectionRepository()
+	err := repo.Save(api_gateway.Connection{
+		ConnectionID: "1",
+		Address:      "address",
+	})
+	assert.NoError(t, err)
+
+	tests := map[string]struct {
+		Input  api_gateway.Connection
+		Output error
+	}{
+		"success": {
+			Input: api_gateway.Connection{
+				ConnectionID: "0",
+				Address:      "address",
+			},
+			Output: nil,
+		},
+		"fail": {
+			Input: api_gateway.Connection{
+				ConnectionID: "1",
+				Address:      "address",
+			},
+			Output: api_gateway.ErrConnectionExists,
+		},
+	}
+
+	for testName, test := range tests {
+		t.Logf("Running '%s' test, caseName: %s", t.Name(), testName)
+		//given
+
+		err := repo.Save(test.Input)
+		if err != nil {
+			assert.Equal(t, test.Output, err)
+			continue
+		}
+
+		c, exist := repo.ConnectionTable[test.Input.ConnectionID]
+		assert.True(t, exist)
+		assert.Equal(t, test.Output, err)
+		assert.Equal(t, test.Input.ConnectionID, c.ConnectionID)
+		assert.Equal(t, test.Input.ConnectionID, c.ConnectionID)
+	}
+
+}
+
+func TestConnectionRepository_Remove(t *testing.T) {
+	repo := api_gateway.NewConnectionRepository()
+	err := repo.Save(api_gateway.Connection{
+		ConnectionID: "0",
+		Address:      "address",
+	})
+	assert.NoError(t, err)
+
+	c, exist := repo.ConnectionTable["0"]
+	assert.True(t, exist)
+
+	repo.Remove(c.ConnectionID)
+	_, exist = repo.ConnectionTable["0"]
+	assert.False(t, exist)
+}
+
+func TestConnectionEventListener_HandleConnectionCreatedEvent(t *testing.T) {
+	repo := api_gateway.NewConnectionRepository()
+	listener := api_gateway.NewConnectionEventListener(*repo)
+
+	err := listener.HandleConnectionCreatedEvent(event.ConnectionCreated{
+		ConnectionID: "0",
+		Address:      "address",
+	})
+	assert.NoError(t, err)
+}
+
+func TestConnectionEventListener_HandleConnectionClosedEvent(t *testing.T) {
+	repo := api_gateway.NewConnectionRepository()
+	listener := api_gateway.NewConnectionEventListener(*repo)
+
+	err := listener.HandleConnectionCreatedEvent(event.ConnectionCreated{
+		ConnectionID: "0",
+		Address:      "address",
+	})
+	assert.NoError(t, err)
+
+	c, exist := repo.ConnectionTable["0"]
+	assert.True(t, exist)
+
+	listener.HandleConnectionClosedEvent(event.ConnectionClosed{
+		ConnectionID: c.ConnectionID,
+	})
+	_, exist = repo.ConnectionTable["0"]
+	assert.False(t, exist)
+}

cmd/connection/dial.go

@@ -54,7 +54,7 @@ func connect(ip string) error {
 			return
 		}
 
-		logger.Infof(nil, "[Cmd] Connection created - Address: [%s], Id:[%s]", connection.Address, connection.ConnectionId)
+		logger.Infof(nil, "[Cmd] Connection created - Address: [%s], Id:[%s]", connection.Address, connection.ConnectionID)
 	})
 
 	if err != nil {

cmd/connection/list.go

@@ -53,7 +53,7 @@ func list() error {
 		fmt.Println("Index\t ID\t\t\t\t\t\t Address")
 		for index, connection := range getConnectionListCommand.ConnectionList {
 			fmt.Printf("[%d]\t [%s]\t [%s]\n",
-				index, connection.ConnectionId, connection.Address)
+				index, connection.ConnectionID, connection.Address)
 		}
 	})
 

cmd/on/grpc_gatewayfx/grpc_gatewayfx.go

@@ -78,7 +78,7 @@ func InitgRPCServer(lifecycle fx.Lifecycle, config *conf.Configuration, hostServ
 		OnStop: func(context context.Context) error {
 			connections, _ := hostService.GetAllConnections()
 			for _, connection := range connections {
-				hostService.CloseConnection(connection.ConnectionId)
+				hostService.CloseConnection(connection.ConnectionID)
 			}
 			hostService.Stop()
 			return nil

common/event/event.go

@@ -168,5 +168,5 @@ type ConnectionCreated struct {
 
 // connection close
 type ConnectionClosed struct {
-	ConnectionId string
+	ConnectionID string
 }

grpc_gateway/api/connection_api.go

@@ -50,20 +50,20 @@ func (c ConnectionApi) CreateConnection(address string) (grpc_gateway.Connection
 		return connection, err
 	}
 
-	logger.Infof(nil, "[gRPC-Gateway] Connection created - Address [%s], ConnectionId [%s]", connection.Address, connection.ConnectionId)
+	logger.Infof(nil, "[gRPC-Gateway] Connection created - Address [%s], ConnectionID [%s]", connection.Address, connection.ConnectionID)
 
 	return connection, nil
 }
 
 func createConnectionCreatedEvent(connection grpc_gateway.Connection) event.ConnectionCreated {
 	return event.ConnectionCreated{
 		Address:      connection.Address,
-		ConnectionID: connection.ConnectionId,
+		ConnectionID: connection.ConnectionID,
 	}
 }
 
 func (c ConnectionApi) CloseConnection(connectionID string) error {
-	logger.Infof(nil, "[gRPC-Gateway] Close connection - ConnectionId [%s]", connectionID)
+	logger.Infof(nil, "[gRPC-Gateway] Close connection - ConnectionID [%s]", connectionID)
 
 	c.grpcService.CloseConnection(connectionID)
 
@@ -72,23 +72,23 @@ func (c ConnectionApi) CloseConnection(connectionID string) error {
 
 func createConnectionClosedEvent(connectionID string) event.ConnectionClosed {
 	return event.ConnectionClosed{
-		ConnectionId: connectionID,
+		ConnectionID: connectionID,
 	}
 }
 
 func (c ConnectionApi) OnConnection(connection grpc_gateway.Connection) {
-	logger.Infof(nil, "[gRPC-Gateway] Connection created - Address [%s], ConnectionId [%s]", connection.Address, connection.ConnectionId)
+	logger.Infof(nil, "[gRPC-Gateway] Connection created - Address [%s], ConnectionID [%s]", connection.Address, connection.ConnectionID)
 
 	if err := c.eventService.Publish("connection.created", createConnectionCreatedEvent(connection)); err != nil {
-		logger.Infof(nil, "[gRPC-Gateway] Fail to publish connection createdEvent - ConnectionId: [%s]", connection.ConnectionId)
+		logger.Infof(nil, "[gRPC-Gateway] Fail to publish connection createdEvent - ConnectionID: [%s]", connection.ConnectionID)
 	}
 }
 
 func (c ConnectionApi) OnDisconnection(connection grpc_gateway.Connection) {
-	logger.Infof(nil, "[gRPC-Gateway] Connection closed - ConnectionId [%s]", connection.ConnectionId)
+	logger.Infof(nil, "[gRPC-Gateway] Connection closed - ConnectionID [%s]", connection.ConnectionID)
 
 	if err := c.eventService.Publish("connection.closed", connection); err != nil {
-		logger.Infof(nil, "[gRPC-Gateway] Fail to publish connection createdEvent - ConnectionId: [%s]", connection.ConnectionId)
+		logger.Infof(nil, "[gRPC-Gateway] Fail to publish connection createdEvent - ConnectionID: [%s]", connection.ConnectionID)
 	}
 }
 

grpc_gateway/connection.go

@@ -17,6 +17,6 @@
 package grpc_gateway
 
 type Connection struct {
-	ConnectionId string
+	ConnectionID string
 	Address      string
 }

grpc_gateway/infra/grpc_service.go

@@ -91,7 +91,7 @@ func (g *GrpcHostService) Dial(address string) (grpc_gateway.Connection, error)
 
 func toGatewayConnectionModel(connection bifrost.Connection) grpc_gateway.Connection {
 	return grpc_gateway.Connection{
-		ConnectionId: connection.GetID(),
+		ConnectionID: connection.GetID(),
 		Address:      connection.GetIP(),
 	}
 }