further codec changes

codec/codec.go

@@ -34,7 +34,7 @@ type Codec interface {
 // the communication, likely followed by the body.
 // In the case of an error, body may be nil.
 type Message struct {
-	Id     uint64
+	Id     string
 	Type   MessageType
 	Target string
 	Method string

codec/jsonrpc/client.go

@@ -19,17 +19,17 @@ type clientCodec struct {
 	resp clientResponse
 
 	sync.Mutex
-	pending map[uint64]string
+	pending map[interface{}]string
 }
 
 type clientRequest struct {
 	Method string         `json:"method"`
 	Params [1]interface{} `json:"params"`
-	ID     uint64         `json:"id"`
+	ID     interface{}    `json:"id"`
 }
 
 type clientResponse struct {
-	ID     uint64           `json:"id"`
+	ID     interface{}      `json:"id"`
 	Result *json.RawMessage `json:"result"`
 	Error  interface{}      `json:"error"`
 }
@@ -39,7 +39,7 @@ func newClientCodec(conn io.ReadWriteCloser) *clientCodec {
 		dec:     json.NewDecoder(conn),
 		enc:     json.NewEncoder(conn),
 		c:       conn,
-		pending: make(map[uint64]string),
+		pending: make(map[interface{}]string),
 	}
 }
 
@@ -71,7 +71,7 @@ func (c *clientCodec) ReadHeader(m *codec.Message) error {
 	c.Unlock()
 
 	m.Error = ""
-	m.Id = c.resp.ID
+	m.Id = fmt.Sprintf("%v", c.resp.ID)
 	if c.resp.Error != nil {
 		x, ok := c.resp.Error.(string)
 		if !ok {

codec/jsonrpc/server.go

@@ -2,9 +2,8 @@ package jsonrpc
 
 import (
 	"encoding/json"
-	"errors"
+	"fmt"
 	"io"
-	"sync"
 
 	"github.com/micro/go-micro/codec"
 )
@@ -17,30 +16,25 @@ type serverCodec struct {
 	// temporary work space
 	req  serverRequest
 	resp serverResponse
-
-	sync.Mutex
-	seq     uint64
-	pending map[uint64]*json.RawMessage
 }
 
 type serverRequest struct {
 	Method string           `json:"method"`
 	Params *json.RawMessage `json:"params"`
-	ID     *json.RawMessage `json:"id"`
+	ID     interface{}      `json:"id"`
 }
 
 type serverResponse struct {
-	ID     *json.RawMessage `json:"id"`
-	Result interface{}      `json:"result"`
-	Error  interface{}      `json:"error"`
+	ID     interface{} `json:"id"`
+	Result interface{} `json:"result"`
+	Error  interface{} `json:"error"`
 }
 
 func newServerCodec(conn io.ReadWriteCloser) *serverCodec {
 	return &serverCodec{
-		dec:     json.NewDecoder(conn),
-		enc:     json.NewEncoder(conn),
-		c:       conn,
-		pending: make(map[uint64]*json.RawMessage),
+		dec: json.NewDecoder(conn),
+		enc: json.NewEncoder(conn),
+		c:   conn,
 	}
 }
 
@@ -50,7 +44,7 @@ func (r *serverRequest) reset() {
 		*r.Params = (*r.Params)[0:0]
 	}
 	if r.ID != nil {
-		*r.ID = (*r.ID)[0:0]
+		r.ID = nil
 	}
 }
 
@@ -60,14 +54,8 @@ func (c *serverCodec) ReadHeader(m *codec.Message) error {
 		return err
 	}
 	m.Method = c.req.Method
-
-	c.Lock()
-	c.seq++
-	c.pending[c.seq] = c.req.ID
+	m.Id = fmt.Sprintf("%v", c.req.ID)
 	c.req.ID = nil
-	m.Id = c.seq
-	c.Unlock()
-
 	return nil
 }
 
@@ -84,19 +72,7 @@ var null = json.RawMessage([]byte("null"))
 
 func (c *serverCodec) Write(m *codec.Message, x interface{}) error {
 	var resp serverResponse
-	c.Lock()
-	b, ok := c.pending[m.Id]
-	if !ok {
-		c.Unlock()
-		return errors.New("invalid sequence number in response")
-	}
-	c.Unlock()
-
-	if b == nil {
-		// Invalid request so no id.  Use JSON null.
-		b = &null
-	}
-	resp.ID = b
+	resp.ID = m.Id
 	resp.Result = x
 	if m.Error == "" {
 		resp.Error = nil

codec/proto/proto.go

@@ -22,11 +22,18 @@ func (c *Codec) ReadBody(b interface{}) error {
 	if err != nil {
 		return err
 	}
+	if b == nil {
+		return nil
+	}
 	return proto.Unmarshal(buf, b.(proto.Message))
 }
 
 func (c *Codec) Write(m *codec.Message, b interface{}) error {
-	buf, err := proto.Marshal(b.(proto.Message))
+	p, ok := b.(proto.Message)
+	if !ok {
+		return nil
+	}
+	buf, err := proto.Marshal(p)
 	if err != nil {
 		return err
 	}

codec/protorpc/protorpc.go

@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"fmt"
 	"io"
+	"strconv"
 	"sync"
 
 	"github.com/golang/protobuf/proto"
@@ -31,13 +32,22 @@ func (c *protoCodec) String() string {
 	return "proto-rpc"
 }
 
+func id(id string) *uint64 {
+	p, err := strconv.ParseInt(id, 10, 64)
+	if err != nil {
+		p = 0
+	}
+	i := uint64(p)
+	return &i
+}
+
 func (c *protoCodec) Write(m *codec.Message, b interface{}) error {
 	switch m.Type {
 	case codec.Request:
 		c.Lock()
 		defer c.Unlock()
 		// This is protobuf, of course we copy it.
-		pbr := &Request{ServiceMethod: &m.Method, Seq: &m.Id}
+		pbr := &Request{ServiceMethod: &m.Method, Seq: id(m.Id)}
 		data, err := proto.Marshal(pbr)
 		if err != nil {
 			return err
@@ -63,7 +73,7 @@ func (c *protoCodec) Write(m *codec.Message, b interface{}) error {
 	case codec.Response:
 		c.Lock()
 		defer c.Unlock()
-		rtmp := &Response{ServiceMethod: &m.Method, Seq: &m.Id, Error: &m.Error}
+		rtmp := &Response{ServiceMethod: &m.Method, Seq: id(m.Id), Error: &m.Error}
 		data, err := proto.Marshal(rtmp)
 		if err != nil {
 			return err
@@ -117,7 +127,7 @@ func (c *protoCodec) ReadHeader(m *codec.Message, mt codec.MessageType) error {
 			return err
 		}
 		m.Method = rtmp.GetServiceMethod()
-		m.Id = rtmp.GetSeq()
+		m.Id = fmt.Sprintf("%d", rtmp.GetSeq())
 	case codec.Response:
 		data, err := ReadNetString(c.rwc)
 		if err != nil {
@@ -129,7 +139,7 @@ func (c *protoCodec) ReadHeader(m *codec.Message, mt codec.MessageType) error {
 			return err
 		}
 		m.Method = rtmp.GetServiceMethod()
-		m.Id = rtmp.GetSeq()
+		m.Id = fmt.Sprintf("%d", rtmp.GetSeq())
 		m.Error = rtmp.GetError()
 	case codec.Publication:
 		_, err := io.Copy(c.buf, c.rwc)