init

Author: sh7ning

.gitignore

@@ -0,0 +1,3 @@
+.idea
+.env
+config.yaml

Dockerfile

@@ -0,0 +1,41 @@
+## 1. 构建应用 ##
+FROM golang:1.15-buster as builder
+
+ARG APP_MODULE="github.com/Gopusher/gateway"
+ARG APP_NAME="gateway"
+
+RUN export GO111MODULE=on \
+    && export GOPROXY=https://goproxy.io \
+    && mkdir -p /go/src/${APP_MODULE}
+
+COPY . /go/src/${APP_MODULE}
+
+# RUN echo /go/src/${APP_MODULE}
+# RUN echo /go/bin/${APP_NAME} ${APP_NAME}.go
+
+RUN cd /go/src/${APP_MODULE} \
+    && CGO_ENABLED=0 go build -ldflags '-s -w' -o /go/bin/${APP_NAME} ./app/${APP_NAME}/app/cmd/main.go
+
+## 2. 应用 ##
+FROM debian:buster
+
+ARG OLD_APP_NAME="gateway"
+ARG APP_NAME="gopusher"
+
+
+# 增加根证书
+RUN apt-get update \
+    && apt-get install ca-certificates -y
+
+COPY --from=builder /go/bin/${OLD_APP_NAME} /usr/local/bin/${APP_NAME}
+
+# .env => /app/.env
+WORKDIR /app
+VOLUME /app
+
+EXPOSE 8900 8901
+
+COPY docker-entrypoint.sh /usr/local/bin/
+ENTRYPOINT ["docker-entrypoint.sh"]
+
+CMD ["${APP_NAME}", "start", "-c", "config.yaml"]

app/gateway/app/api/anycall.go

@@ -0,0 +1,29 @@
+package api
+
+import (
+	"encoding/json"
+)
+
+type AnyCallMessage struct {
+	TokenMessage
+	Method string          `json:"method"`
+	Args   json.RawMessage `json:"args"`
+}
+
+//获取部分 order book
+func (s *Server) AnyCall(message *AnyCallMessage, reply *Response) error {
+	if errResp := s.checkToken(message.Token); errResp != nil {
+		*reply = *errResp
+		return nil
+	}
+	//log.Debug("AnyCall method: " + message.Method + ", args: " + string(message.Args))
+
+	data, err := s.server.AnyCall(message.Method, message.Args)
+	if err != nil {
+		*reply = s.failure(ServerErrorCode, err.Error())
+		return nil
+	}
+
+	*reply = s.success(data)
+	return nil
+}

app/gateway/app/api/api.go

@@ -0,0 +1,121 @@
+package api
+
+import (
+	"fmt"
+	"net"
+	"net/rpc"
+	"net/rpc/jsonrpc"
+
+	"github.com/gopusher/gateway/app/gateway/app/protocols"
+	"github.com/gopusher/gateway/pkg/log"
+	"go.uber.org/zap"
+)
+
+type Config struct {
+	Address string `mapstructure:"address" validate:"required"`
+	Token   string `mapstructure:"token"`
+}
+
+//Server is api server
+type Server struct {
+	node   string
+	server protocols.Server
+	config *Config
+}
+
+//InitRpcServer init rpc server
+func InitRpcServer(node string, server protocols.Server, config *Config) {
+	if err := rpc.Register(&Server{
+		node:   node,
+		server: server,
+		config: config,
+	}); err != nil {
+		log.Panic("Gateway api server run failed, error: "+err.Error(), zap.Error(err))
+	}
+
+	listener, err := net.Listen("tcp", config.Address)
+	if err != nil {
+		log.Panic("Gateway api server run failed, error: "+err.Error(), zap.Error(err))
+	}
+
+	log.Info(fmt.Sprintf(
+		"Gateway api server start running, Node: %s, gateway api address: %s, token: %s",
+		node, config.Address, config.Token,
+	))
+	defer func() {
+		if err := listener.Close(); err != nil {
+			log.Panic("Gateway api server run failed, error: "+err.Error(), zap.Error(err))
+		}
+	}()
+	for {
+		conn, err := listener.Accept()
+		if err != nil {
+			continue
+		}
+
+		go jsonrpc.ServeConn(conn)
+	}
+}
+
+//TokenMessage is token type message
+type TokenMessage struct {
+	Token string `json:"token"` //作为消息发送鉴权
+}
+
+//ConnectionsMessage is a connection type message
+type ConnectionsMessage struct {
+	Connections []string `json:"connections"` //消息接受者
+	TokenMessage
+}
+
+//Response is api response
+type Response struct {
+	Code  int         `json:"code"`
+	Data  interface{} `json:"data"`
+	Error string      `json:"error"`
+}
+
+const (
+	SuccessCode = 20000
+
+	ParametersError = 40000
+
+	TokenErrorCode = 40100
+
+	ServerErrorCode = 50000
+
+	SendToConnectionsErrorCode = 50001
+)
+
+func (s *Server) success(data interface{}) Response {
+	return Response{
+		Code:  SuccessCode,
+		Data:  data,
+		Error: "",
+	}
+}
+
+func (s *Server) failure(code int, err string) Response {
+	return Response{
+		Code:  code,
+		Data:  "",
+		Error: err,
+	}
+}
+
+func (s *Server) failureWithData(code int, data interface{}, err string) Response {
+	return Response{
+		Code:  code,
+		Data:  data,
+		Error: err,
+	}
+}
+
+func (s *Server) checkToken(token string) *Response {
+	if token != s.config.Token {
+		resp := s.failure(TokenErrorCode, "error rpc token")
+		return &resp
+	}
+
+	return nil
+}

app/gateway/app/api/check_online.go

@@ -0,0 +1,30 @@
+package api
+
+//CheckConnectionsOnline check connections is online
+func (s *Server) CheckConnectionsOnline(message *ConnectionsMessage, reply *Response) error {
+	if errResp := s.checkToken(message.Token); errResp != nil {
+		*reply = *errResp
+		return nil
+	}
+
+	var onlineConnections []string
+	if len(message.Connections) > 0 {
+		onlineConnections = s.server.CheckConnectionsOnline(message.Connections)
+	}
+
+	*reply = s.success(onlineConnections)
+	return nil
+}
+
+//GetAllConnections returns all connections
+func (s *Server) GetAllConnections(message *TokenMessage, reply *Response) error {
+	if errResp := s.checkToken(message.Token); errResp != nil {
+		*reply = *errResp
+		return nil
+	}
+
+	onlineConnections := s.server.GetAllConnections()
+
+	*reply = s.success(onlineConnections)
+	return nil
+}

app/gateway/app/api/client/example.go

@@ -0,0 +1,56 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"net/rpc/jsonrpc"
+	"os"
+)
+
+type message struct {
+	Connections []string `json:"connections"` //消息接受者
+	Msg         string   `json:"msg"`         //为一个json，里边包含 type 消息类型
+	Token       string   `json:"token"`       //作为消息发送鉴权
+}
+
+type kickMessage struct {
+	Connections []string `json:"connections"` //消息接受者
+	Token       string   `json:"token"`       //作为消息发送鉴权
+}
+
+func main() {
+	if len(os.Args) < 2 {
+		log.Fatal("消息发送参数格式: go run ./example.go msg ...to")
+	}
+
+	//连接远程rpc服务
+	//这里使用jsonrpc.Dial
+	//todo 这里的 ip 要注意
+	rpc, err := jsonrpc.Dial("tcp", "192.168.3.165:8901")
+	if err != nil {
+		log.Fatal(err)
+	}
+	//response 为 json 字符串
+	var response string
+	//调用远程方法
+	//注意第三个参数是指针类型
+
+	//发送消息
+	err2 := rpc.Call("Server.SendToConnections", &message{
+		Connections: os.Args[2:],
+		Msg:         os.Args[1],
+		Token:       "token",
+	}, &response)
+
+	//kick conns
+	//err2 := rpc.Call("Server.KickConnections", &kickMessage{
+	//	Connections: os.Args[1:],
+	//	Token: "token",
+	//}, &response)
+
+	if err2 != nil {
+		log.Fatal(err2)
+	}
+
+	fmt.Println(response)
+}

app/gateway/app/api/client/example.py

@@ -0,0 +1,92 @@
+# -*- coding: UTF-8 -*-
+import json
+import socket
+
+
+class Rpc(object):
+    host: str
+    port: int
+    token: str
+    conn: socket.socket
+
+    def __init__(self, host, port, token):
+        self.host = host
+        self.port = port
+        self.token = token
+        # self.conn = self.get_connect()
+
+    def connect(self):
+        if self.conn is None:
+            self.conn = socket.create_connection((self.host, self.port), timeout=1.5)
+        return self.conn
+
+    def __getattr__(self, method):
+        if method in ["conn"]:
+            return None
+
+        def func(**kwargs):
+            # args = kwargs if len(kwargs) else args
+            params = {
+                'token': self.token,
+            }
+            if kwargs is not None:
+                params.update(kwargs)
+
+            data = {
+                'method': "Server." + method,
+                'params': [params],
+            }
+
+            return self.execute(data)
+
+        return func
+
+    def execute(self, data):
+        self.connect()
+
+        msg_id = 0
+        data['id'] = msg_id
+        msg = json.dumps(data)
+        self.conn.sendall(msg.encode())
+
+        resp = self.read_line()
+        if not resp:
+            self.close()
+            raise Exception("rpc 获取数据失败, Not resp, server gone")
+
+        resp = json.loads(resp)
+
+        if resp.get('id') != msg_id:
+            raise Exception("expected id=%s, received id=%s: %s"
+                            % (msg_id, resp.get('id'), resp.get('error')))
+
+        if resp.get('error') is not None:
+            raise Exception(resp.get('error'))
+
+        data = json.loads(resp.get('result'))
+        if data['code'] != '0':
+            raise Exception("rpc 获取数据失败: %s" % data['error'])
+
+        return data
+
+    def read_line(self):
+        # return self.conn.makefile().readline()
+
+        ret = b''
+        while True:
+            c = self.conn.recv(1)
+            if c == b'\n' or c == b'':
+                break
+            else:
+                ret += c
+
+        return ret.decode("utf-8")
+
+    def close(self):
+        self.conn.close()
+        self.conn = None
+
+
+if __name__ == '__main__':
+    rpc = Rpc('127.0.0.1', 8901, 'token')
+    print(rpc.SendToConnections(connections=['a', 'b']))