ing

Gopkg.lock

@@ -16,8 +16,12 @@
 [[projects]]
   branch = "master"
   name = "git.loafle.net/commons/server-go"
-  packages = ["client"]
+  packages = [
-  revision = "1c6419e15b8c2c9ca609d205bf907948a19c6f1d"
+    ".",
+    "socket",
+    "socket/client"
+  ]
+  revision = "d1d485c711ef8c59ad51fbf17dba80158f1a8ace"
 
 [[projects]]
   branch = "master"
@@ -70,6 +74,6 @@
 [solve-meta]
   analyzer-name = "dep"
   analyzer-version = 1
-  inputs-digest = "c437dcb9b279fd0a7fe610d8cce5eb7e88bb92350b0571ade6285479a6ef0413"
+  inputs-digest = "5b81e7a1b9b142f7de01bacc57b6a7be2599965fd84b170a42fb1ef090910bcc"
   solver-name = "gps-cdcl"
   solver-version = 1

client/client-handler.go

@@ -8,7 +8,7 @@ import (
 	"git.loafle.net/commons/rpc-go"
 	"git.loafle.net/commons/rpc-go/protocol"
 	"git.loafle.net/commons/rpc-go/registry"
-	csc "git.loafle.net/commons/server-go/client"
+	csc "git.loafle.net/commons/server-go/socket/client"
 )
 
 type ClientHandler interface {

client/client.go

@@ -9,7 +9,8 @@ import (
 
 	"git.loafle.net/commons/logging-go"
 	"git.loafle.net/commons/rpc-go/protocol"
-	csc "git.loafle.net/commons/server-go/client"
+	css "git.loafle.net/commons/server-go/socket"
+	cssc "git.loafle.net/commons/server-go/socket/client"
 )
 
 var uint64Type = reflect.TypeOf(uint64(0))
@@ -17,7 +18,7 @@ var uint64Type = reflect.TypeOf(uint64(0))
 type Client struct {
 	ClientHandler ClientHandler
 
-	ctx      csc.ClientCtx
+	ctx      cssc.ClientCtx
 	stopChan chan struct{}
 	stopWg   sync.WaitGroup
 
@@ -159,7 +160,7 @@ func (c *Client) internalSend(hasResponse bool, result interface{}, method strin
 	}
 }
 
-func (c *Client) handleClient(readChan <-chan []byte, writeChan chan<- []byte) {
+func (c *Client) handleClient(readChan <-chan css.SocketMessage, writeChan chan<- css.SocketMessage) {
 	defer func() {
 
 		if err := c.ClientHandler.GetConnector().Disconnect(); nil != err {
@@ -199,7 +200,7 @@ func (c *Client) handleClient(readChan <-chan []byte, writeChan chan<- []byte) {
 	}
 }
 
-func (c *Client) handleSend(stopChan <-chan struct{}, doneChan chan<- error, writeChan chan<- []byte) {
+func (c *Client) handleSend(stopChan <-chan struct{}, doneChan chan<- error, writeChan chan<- css.SocketMessage) {
 	var (
 		rs      *requestState
 		id      interface{}
@@ -240,7 +241,7 @@ LOOP:
 			}
 
 			select {
-			case writeChan <- message:
+			case writeChan <- css.MakeSocketMessage(css.TextMessage, message):
 			default:
 				rs.setError(fmt.Errorf("%s cannot send request", c.logHeader()))
 				rs.done()
@@ -256,8 +257,10 @@ LOOP:
 	}
 }
 
-func (c *Client) handleReceive(stopChan <-chan struct{}, doneChan chan<- error, readChan <-chan []byte) {
+func (c *Client) handleReceive(stopChan <-chan struct{}, doneChan chan<- error, readChan <-chan css.SocketMessage) {
 	var (
+		socketMessage css.SocketMessage
+		messageType   int
 		message       []byte
 		err           error
 		ok            bool
@@ -270,11 +273,14 @@ func (c *Client) handleReceive(stopChan <-chan struct{}, doneChan chan<- error,
 LOOP:
 	for {
 		select {
-		case message, ok = <-readChan:
+		case socketMessage, ok = <-readChan:
 			if !ok {
 				return
 			}
-			resCodec, err := c.ClientHandler.GetRPCCodec().NewResponse(message)
+
+			messageType, message = socketMessage()
+
+			resCodec, err := c.ClientHandler.GetRPCCodec().NewResponse(messageType, message)
 			if nil != err {
 				logging.Logger().Debug(err)
 				continue LOOP

codec/codec.go

@@ -0,0 +1,44 @@
+package codec
+
+import (
+	css "git.loafle.net/commons/server-go/socket"
+)
+
+type Codec interface {
+	Encode(message []byte) ([]byte, error)
+	Decode(message []byte) ([]byte, error)
+}
+
+type codec struct {
+}
+
+func (_ *codec) Encode(message []byte) ([]byte, error) {
+	return message, nil
+}
+
+func (_ *codec) Decode(message []byte) ([]byte, error) {
+	return message, nil
+}
+
+var DefaultCodec = &codec{}
+
+type CodecSelector interface {
+	Encode(message []byte) (messageType int, encMessage []byte, err error)
+	Decode(messageType int, message []byte) (decMessage []byte, err error)
+}
+
+type codecSelector struct {
+}
+
+func (_ *codecSelector) Encode(message []byte) (messageType int, encMessage []byte, err error) {
+	messageType = css.TextMessage
+	encMessage, err = DefaultCodec.Encode(message)
+	return
+}
+
+func (_ *codecSelector) Decode(messageType int, message []byte) (decMessage []byte, err error) {
+	decMessage, err = DefaultCodec.Decode(message)
+	return
+}
+
+var DefaultCodecSelector = &codecSelector{}

codec/compression_codec.go

@@ -0,0 +1,77 @@
+package codec
+
+import (
+	"bytes"
+	"compress/gzip"
+	"io/ioutil"
+
+	css "git.loafle.net/commons/server-go/socket"
+)
+
+type GZipCodec struct {
+}
+
+func (_ *GZipCodec) Encode(message []byte) ([]byte, error) {
+	var b bytes.Buffer
+	gWriter, err := gzip.NewWriterLevel(&b, gzip.BestSpeed)
+	if nil != err {
+		return nil, err
+	}
+
+	if _, err := gWriter.Write(message); nil != err {
+		return nil, err
+	}
+	if err := gWriter.Flush(); nil != err {
+		return nil, err
+	}
+	if err := gWriter.Close(); nil != err {
+		return nil, err
+	}
+
+	return b.Bytes(), nil
+}
+
+func (_ *GZipCodec) Decode(message []byte) ([]byte, error) {
+	gReader, err := gzip.NewReader(bytes.NewBuffer(message))
+	if nil != err {
+		return nil, err
+	}
+	defer gReader.Close()
+
+	b, err := ioutil.ReadAll(gReader)
+	if nil != err {
+		return nil, err
+	}
+
+	return b, nil
+}
+
+var gZipCodec = &GZipCodec{}
+
+type compressionCodecSelector struct {
+	threshold int
+}
+
+func (cs *compressionCodecSelector) Encode(message []byte) (messageType int, encMessage []byte, err error) {
+	switch {
+	case cs.threshold < len(message):
+		messageType = css.BinaryMessage
+		encMessage, err = gZipCodec.Encode(message)
+	default:
+		messageType = css.TextMessage
+		encMessage, err = DefaultCodec.Encode(message)
+	}
+
+	return
+}
+
+func (cs *compressionCodecSelector) Decode(messageType int, message []byte) (decMessage []byte, err error) {
+	switch messageType {
+	case css.BinaryMessage:
+		decMessage, err = gZipCodec.Decode(message)
+	default:
+		decMessage, err = DefaultCodec.Decode(message)
+	}
+
+	return
+}

protocol/client_codec.go

@@ -2,8 +2,8 @@ package protocol
 
 // ClientCodec creates a ClientCodecRequest to process each request.
 type ClientCodec interface {
-	NewRequest(method string, args []interface{}, id interface{}) ([]byte, error)
+	NewRequest(method string, args []interface{}, id interface{}) (messageType int, message []byte, err error)
-	NewResponse(buf []byte) (ClientResponseCodec, error)
+	NewResponse(messageType int, message []byte) (ClientResponseCodec, error)
 }
 
 type ClientResponseCodec interface {

protocol/json/client.go

@@ -3,6 +3,7 @@ package json
 import (
 	"encoding/json"
 
+	"git.loafle.net/commons/rpc-go/codec"
 	"git.loafle.net/commons/rpc-go/protocol"
 )
 
@@ -10,19 +11,27 @@ import (
 // Codec
 // ----------------------------------------------------------------------------
 
+// NewClientCodec returns a new JSON Codec.
+func NewCustomClientCodec(codecSelector codec.CodecSelector) protocol.ClientCodec {
+	return &ClientCodec{
+		codecSelector: codecSelector,
+	}
+}
+
 // NewClientCodec returns a new JSON Codec.
 func NewClientCodec() protocol.ClientCodec {
-	return &ClientCodec{}
+	return NewCustomClientCodec(codec.DefaultCodecSelector)
 }
 
 // ClientCodec creates a ClientCodecRequest to process each request.
 type ClientCodec struct {
+	codecSelector codec.CodecSelector
 }
 
-func (cc *ClientCodec) NewRequest(method string, args []interface{}, id interface{}) ([]byte, error) {
+func (cc *ClientCodec) NewRequest(method string, args []interface{}, id interface{}) (messageType int, message []byte, err error) {
 	params, err := convertParamsToStringArray(args)
 	if nil != err {
-		return nil, err
+		return 0, nil, err
 	}
 
 	req := &clientRequest{
@@ -32,9 +41,19 @@ func (cc *ClientCodec) NewRequest(method string, args []interface{}, id interfac
 		ID:      id,
 	}
 
-	return json.Marshal(req)
+	buf, err := json.Marshal(req)
+	if nil != err {
+		return 0, nil, err
+	}
+
+	return cc.codecSelector.Encode(buf)
 }
 
-func (cc *ClientCodec) NewResponse(buf []byte) (protocol.ClientResponseCodec, error) {
+func (cc *ClientCodec) NewResponse(messageType int, message []byte) (protocol.ClientResponseCodec, error) {
+	buf, err := cc.codecSelector.Decode(messageType, message)
+	if nil != err {
+		return nil, err
+	}
+
 	return newClientResponseCodec(buf)
 }

protocol/json/server.go

@@ -3,6 +3,7 @@ package json
 import (
 	"encoding/json"
 
+	"git.loafle.net/commons/rpc-go/codec"
 	"git.loafle.net/commons/rpc-go/protocol"
 )
 
@@ -12,43 +13,65 @@ var null = json.RawMessage([]byte("null"))
 // Codec
 // ----------------------------------------------------------------------------
 
+// NewCustomServerCodec returns a new JSON Codec.
+func NewCustomServerCodec(codecSelector codec.CodecSelector) protocol.ServerCodec {
+	return &ServerCodec{
+		codecSelector: codecSelector,
+	}
+}
+
 // NewServerCodec returns a new JSON Codec.
 func NewServerCodec() protocol.ServerCodec {
-	return &ServerCodec{}
+	return NewCustomServerCodec(codec.DefaultCodecSelector)
 }
 
 // ServerCodec creates a ServerRequestCodec to process each request.
 type ServerCodec struct {
+	codecSelector codec.CodecSelector
 }
 
-func (sc *ServerCodec) NewRequest(buf []byte) (protocol.ServerRequestCodec, error) {
+func (sc *ServerCodec) NewRequest(messageType int, message []byte) (protocol.ServerRequestCodec, error) {
-	return newServerRequestCodec(buf)
+	buf, err := sc.codecSelector.Decode(messageType, message)
-}
-
-func (sc *ServerCodec) NewRequestWithString(method string, params []string, id interface{}) (protocol.ServerRequestCodec, error) {
-	req := &clientRequest{
-		Version: Version,
-		Method:  method,
-		Params:  params,
-		ID:      id,
-	}
-
-	buf, err := json.Marshal(req)
 	if nil != err {
 		return nil, err
 	}
-
+	return newServerRequestCodec(sc.codecSelector, buf)
-	return sc.NewRequest(buf)
 }
 
-func (sc *ServerCodec) NewNotification(method string, args []interface{}) ([]byte, error) {
+// func (sc *ServerCodec) NewRequestWithString(method string, params []string, id interface{}) (protocol.ServerRequestCodec, error) {
+// 	req := &clientRequest{
+// 		Version: Version,
+// 		Method:  method,
+// 		Params:  params,
+// 		ID:      id,
+// 	}
+
+// 	buf, err := json.Marshal(req)
+// 	if nil != err {
+// 		return nil, err
+// 	}
+
+// 	return sc.NewRequest(buf)
+// }
+
+func (sc *ServerCodec) NewNotification(method string, args []interface{}) (messageType int, message []byte, err error) {
 	params, err := convertParamsToStringArray(args)
 	if nil != err {
-		return nil, err
+		return 0, nil, err
 	}
 
 	noti := &serverNotification{Method: method, Params: params}
 	res := &serverResponse{Version: Version, Result: noti}
 
-	return json.Marshal(res)
+	buf, err := json.Marshal(res)
+	if nil != err {
+		return 0, nil, err
+	}
+
+	messageType, message, err = sc.codecSelector.Encode(buf)
+	if nil != err {
+		return 0, nil, err
+	}
+
+	return
 }

protocol/json/server_request.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"strings"
 
+	"git.loafle.net/commons/rpc-go/codec"
 	"git.loafle.net/commons/rpc-go/protocol"
 	cuej "git.loafle.net/commons/util-go/encoding/json"
 )
@@ -34,7 +35,7 @@ type serverRequest struct {
 // ----------------------------------------------------------------------------
 
 // newRequestCodec returns a new ServerRequestCodec.
-func newServerRequestCodec(buf []byte) (protocol.ServerRequestCodec, error) {
+func newServerRequestCodec(codecSelector codec.CodecSelector, buf []byte) (protocol.ServerRequestCodec, error) {
 
 	req := &serverRequest{}
 	err := json.Unmarshal(buf, req)
@@ -55,11 +56,12 @@ func newServerRequestCodec(buf []byte) (protocol.ServerRequestCodec, error) {
 		}
 	}
 
-	return &ServerRequestCodec{req: req, err: err}, nil
+	return &ServerRequestCodec{codecSelector: codecSelector, req: req, err: err}, nil
 }
 
 // ServerRequestCodec decodes and encodes a single request.
 type ServerRequestCodec struct {
+	codecSelector codec.CodecSelector
 	req           *serverRequest
 	err           error
 }
@@ -123,25 +125,34 @@ func (src *ServerRequestCodec) Params() ([]string, error) {
 	return nil, src.err
 }
 
-func (src *ServerRequestCodec) NewResponse(reply interface{}, err error) ([]byte, error) {
+func (src *ServerRequestCodec) encodeResponse(res *serverResponse) (messageType int, message []byte, err error) {
-	res := src.newServerResponse(reply, err)
+	buf, err := json.Marshal(res)
-	return json.Marshal(res)
+	if nil != err {
+		return 0, nil, err
+	}
+
+	return src.codecSelector.Encode(buf)
 }
 
-func (src *ServerRequestCodec) NewResponseWithString(reply string, err error) ([]byte, error) {
+func (src *ServerRequestCodec) NewResponse(reply interface{}, replyErr error) (messageType int, message []byte, err error) {
-	res := src.newServerResponse(nil, err)
+	res := src.newServerResponse(reply, replyErr)
+	return src.encodeResponse(res)
+}
+
+func (src *ServerRequestCodec) NewResponseWithString(reply string, replyErr error) (messageType int, message []byte, err error) {
+	res := src.newServerResponse(nil, replyErr)
 
 	r := strings.TrimSpace(reply)
 	if nil == err && "" != r {
 		var i interface{}
 		mErr := json.Unmarshal([]byte(r), &i)
 		if nil != mErr {
-			return nil, mErr
+			return 0, nil, mErr
 		}
 
 		res.Result = i
 	}
-	return json.Marshal(res)
+	return src.encodeResponse(res)
 }
 
 func (src *ServerRequestCodec) newServerResponse(reply interface{}, err error) *serverResponse {

protocol/server_codec.go

@@ -2,9 +2,9 @@ package protocol
 
 // ServerCodec creates a ServerRequestCodec to process each request.
 type ServerCodec interface {
-	NewRequest(buf []byte) (ServerRequestCodec, error)
+	NewRequest(messageType int, message []byte) (ServerRequestCodec, error)
-	NewRequestWithString(method string, params []string, id interface{}) (ServerRequestCodec, error)
+	// NewRequestWithString(method string, params []string, id interface{}) (ServerRequestCodec, error)
-	NewNotification(method string, args []interface{}) ([]byte, error)
+	NewNotification(method string, args []interface{}) (messageType int, message []byte, err error)
 }
 
 // ServerRequestCodec decodes a request and encodes a response using a specific
@@ -13,6 +13,6 @@ type ServerRequestCodec interface {
 	RegistryCodec
 
 	HasResponse() bool
-	NewResponse(reply interface{}, err error) ([]byte, error)
+	NewResponse(reply interface{}, replyErr error) (messageType int, message []byte, err error)
-	NewResponseWithString(reply string, err error) ([]byte, error)
+	NewResponseWithString(reply string, replyErr error) (messageType int, message []byte, err error)
 }