unistack-org/micro
4
0
Fork 2
Code Issues 18 Pull Requests Actions Packages Projects Releases 331 Wiki Activity

many fixes for lint and context.Context usage (#5)

Browse Source 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов
2020-11-03 02:02:32 +03:00
committed by GitHub
parent 40b0870cf8
commit 8a2b122015
44 changed files with 152 additions and 1175 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
codec/codec.go
View File

@@ -14,12 +14,14 @@ const (
)
var (
// ErrInvalidMessage returned when invalid messge passed to codec
ErrInvalidMessage = errors.New("invalid message")
)
// MessageType
type MessageType int
// Takes in a connection/buffer and returns a new Codec
// NewCodec takes in a connection/buffer and returns a new Codec
type NewCodec func(io.ReadWriteCloser) Codec
// Codec encodes/decodes various types of messages used within go-micro.
@@ -34,11 +36,13 @@ type Codec interface {
String() string
}
// Reader interface
type Reader interface {
ReadHeader(*Message, MessageType) error
ReadBody(interface{}) error
}
// Writer interface
type Writer interface {
Write(*Message, interface{}) error
}

60
codec/json/json.go
View File

@@ -1,60 +0,0 @@
// Package json provides a json codec
package json
import (
"encoding/json"
"io"
"io/ioutil"
"github.com/unistack-org/micro/v3/codec"
jsonpb "google.golang.org/protobuf/encoding/protojson"
"google.golang.org/protobuf/proto"
)
type Codec struct {
Conn io.ReadWriteCloser
Encoder *json.Encoder
Decoder *json.Decoder
}
func (c *Codec) ReadHeader(m *codec.Message, t codec.MessageType) error {
return nil
}
func (c *Codec) ReadBody(b interface{}) error {
if b == nil {
return nil
}
switch m := b.(type) {
case proto.Message:
buf, err := ioutil.ReadAll(c.Conn)
if err != nil {
return err
}
return jsonpb.Unmarshal(buf, m)
}
return c.Decoder.Decode(b)
}
func (c *Codec) Write(m *codec.Message, b interface{}) error {
if b == nil {
return nil
}
return c.Encoder.Encode(b)
}
func (c *Codec) Close() error {
return c.Conn.Close()
}
func (c *Codec) String() string {
return "json"
}
func NewCodec(c io.ReadWriteCloser) codec.Codec {
return &Codec{
Conn: c,
Decoder: json.NewDecoder(c),
Encoder: json.NewEncoder(c),
}
}

45
codec/json/marshaler.go
View File

@@ -1,45 +0,0 @@
package json
import (
"bytes"
"encoding/json"
oldjsonpb "github.com/golang/protobuf/jsonpb"
oldproto "github.com/golang/protobuf/proto"
"github.com/oxtoacart/bpool"
jsonpb "google.golang.org/protobuf/encoding/protojson"
"google.golang.org/protobuf/proto"
)
var jsonpbMarshaler = &jsonpb.MarshalOptions{}
var oldjsonpbMarshaler = &oldjsonpb.Marshaler{}
// create buffer pool with 16 instances each preallocated with 256 bytes
var bufferPool = bpool.NewSizedBufferPool(16, 256)
type Marshaler struct{}
func (j Marshaler) Marshal(v interface{}) ([]byte, error) {
switch m := v.(type) {
case proto.Message:
return jsonpbMarshaler.Marshal(m)
case oldproto.Message:
buf, err := oldjsonpbMarshaler.MarshalToString(m)
return []byte(buf), err
}
return json.Marshal(v)
}
func (j Marshaler) Unmarshal(d []byte, v interface{}) error {
switch m := v.(type) {
case proto.Message:
return jsonpb.Unmarshal(d, m)
case oldproto.Message:
return oldjsonpb.Unmarshal(bytes.NewReader(d), m)
}
return json.Unmarshal(d, v)
}
func (j Marshaler) String() string {
return "json"
}

97
codec/jsonrpc/client.go
View File

@@ -1,97 +0,0 @@
package jsonrpc
import (
"encoding/json"
"fmt"
"io"
"sync"
"github.com/unistack-org/micro/v3/codec"
)
type clientCodec struct {
dec *json.Decoder // for reading JSON values
enc *json.Encoder // for writing JSON values
c io.Closer
// temporary work space
req clientRequest
resp clientResponse
sync.Mutex
pending map[interface{}]string
}
type clientRequest struct {
Method string `json:"method"`
Params [1]interface{} `json:"params"`
ID interface{} `json:"id"`
}
type clientResponse struct {
ID interface{} `json:"id"`
Result *json.RawMessage `json:"result"`
Error interface{} `json:"error"`
}
func newClientCodec(conn io.ReadWriteCloser) *clientCodec {
return &clientCodec{
dec: json.NewDecoder(conn),
enc: json.NewEncoder(conn),
c: conn,
pending: make(map[interface{}]string),
}
}
func (c *clientCodec) Write(m *codec.Message, b interface{}) error {
c.Lock()
c.pending[m.Id] = m.Method
c.Unlock()
c.req.Method = m.Method
c.req.Params[0] = b
c.req.ID = m.Id
return c.enc.Encode(&c.req)
}
func (r *clientResponse) reset() {
r.ID = 0
r.Result = nil
r.Error = nil
}
func (c *clientCodec) ReadHeader(m *codec.Message) error {
c.resp.reset()
if err := c.dec.Decode(&c.resp); err != nil {
return err
}
c.Lock()
m.Method = c.pending[c.resp.ID]
delete(c.pending, c.resp.ID)
c.Unlock()
m.Error = ""
m.Id = fmt.Sprintf("%v", c.resp.ID)
if c.resp.Error != nil {
x, ok := c.resp.Error.(string)
if !ok {
return fmt.Errorf("invalid error %v", c.resp.Error)
}
if x == "" {
x = "unspecified error"
}
m.Error = x
}
return nil
}
func (c *clientCodec) ReadBody(x interface{}) error {
if x == nil || c.resp.Result == nil {
return nil
}
return json.Unmarshal(*c.resp.Result, x)
}
func (c *clientCodec) Close() error {
return c.c.Close()
}

88
codec/jsonrpc/jsonrpc.go
View File

@@ -1,88 +0,0 @@
// Package jsonrpc provides a json-rpc 1.0 codec
package jsonrpc
import (
"bytes"
"encoding/json"
"fmt"
"io"
"github.com/unistack-org/micro/v3/codec"
)
type jsonCodec struct {
buf *bytes.Buffer
mt codec.MessageType
rwc io.ReadWriteCloser
c *clientCodec
s *serverCodec
}
func (j *jsonCodec) Close() error {
j.buf.Reset()
return j.rwc.Close()
}
func (j *jsonCodec) String() string {
return "json-rpc"
}
func (j *jsonCodec) Write(m *codec.Message, b interface{}) error {
switch m.Type {
case codec.Request:
return j.c.Write(m, b)
case codec.Response, codec.Error:
return j.s.Write(m, b)
case codec.Event:
data, err := json.Marshal(b)
if err != nil {
return err
}
_, err = j.rwc.Write(data)
return err
default:
return fmt.Errorf("Unrecognised message type: %v", m.Type)
}
}
func (j *jsonCodec) ReadHeader(m *codec.Message, mt codec.MessageType) error {
j.buf.Reset()
j.mt = mt
switch mt {
case codec.Request:
return j.s.ReadHeader(m)
case codec.Response:
return j.c.ReadHeader(m)
case codec.Event:
_, err := io.Copy(j.buf, j.rwc)
return err
default:
return fmt.Errorf("Unrecognised message type: %v", mt)
}
}
func (j *jsonCodec) ReadBody(b interface{}) error {
switch j.mt {
case codec.Request:
return j.s.ReadBody(b)
case codec.Response:
return j.c.ReadBody(b)
case codec.Event:
if b != nil {
return json.Unmarshal(j.buf.Bytes(), b)
}
default:
return fmt.Errorf("Unrecognised message type: %v", j.mt)
}
return nil
}
func NewCodec(rwc io.ReadWriteCloser) codec.Codec {
return &jsonCodec{
buf: bytes.NewBuffer(nil),
rwc: rwc,
c: newClientCodec(rwc),
s: newServerCodec(rwc),
}
}

84
codec/jsonrpc/server.go
View File

@@ -1,84 +0,0 @@
package jsonrpc
import (
"encoding/json"
"fmt"
"io"
"github.com/unistack-org/micro/v3/codec"
)
type serverCodec struct {
dec *json.Decoder // for reading JSON values
enc *json.Encoder // for writing JSON values
c io.Closer
// temporary work space
req serverRequest
}
type serverRequest struct {
Method string `json:"method"`
Params *json.RawMessage `json:"params"`
ID interface{} `json:"id"`
}
type serverResponse struct {
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,
}
}
func (r *serverRequest) reset() {
r.Method = ""
if r.Params != nil {
*r.Params = (*r.Params)[0:0]
}
if r.ID != nil {
r.ID = nil
}
}
func (c *serverCodec) ReadHeader(m *codec.Message) error {
c.req.reset()
if err := c.dec.Decode(&c.req); err != nil {
return err
}
m.Method = c.req.Method
m.Id = fmt.Sprintf("%v", c.req.ID)
c.req.ID = nil
return nil
}
func (c *serverCodec) ReadBody(x interface{}) error {
if x == nil {
return nil
}
var params [1]interface{}
params[0] = x
return json.Unmarshal(*c.req.Params, &params)
}
func (c *serverCodec) Write(m *codec.Message, x interface{}) error {
var resp serverResponse
resp.ID = m.Id
resp.Result = x
if m.Error == "" {
resp.Error = nil
} else {
resp.Error = m.Error
}
return c.enc.Encode(resp)
}
func (c *serverCodec) Close() error {
return c.c.Close()
}

47
codec/proto/marshaler.go
View File

@@ -1,47 +0,0 @@
package proto
import (
"bytes"
"github.com/golang/protobuf/proto"
"github.com/oxtoacart/bpool"
"github.com/unistack-org/micro/v3/codec"
)
// create buffer pool with 16 instances each preallocated with 256 bytes
var bufferPool = bpool.NewSizedBufferPool(16, 256)
type Marshaler struct{}
func (Marshaler) Marshal(v interface{}) ([]byte, error) {
pb, ok := v.(proto.Message)
if !ok {
return nil, codec.ErrInvalidMessage
}
// looks not good, but allows to reuse underlining bytes
buf := bufferPool.Get()
pbuf := proto.NewBuffer(buf.Bytes())
defer func() {
bufferPool.Put(bytes.NewBuffer(pbuf.Bytes()))
}()
if err := pbuf.Marshal(pb); err != nil {
return nil, err
}
return pbuf.Bytes(), nil
}
func (Marshaler) Unmarshal(data []byte, v interface{}) error {
pb, ok := v.(proto.Message)
if !ok {
return codec.ErrInvalidMessage
}
return proto.Unmarshal(data, pb)
}
func (Marshaler) String() string {
return "proto"
}

37
codec/proto/message.go
View File

@@ -1,37 +0,0 @@
package proto
type Message struct {
Data []byte
}
func (m *Message) MarshalJSON() ([]byte, error) {
return m.Data, nil
}
func (m *Message) UnmarshalJSON(data []byte) error {
m.Data = data
return nil
}
func (m *Message) ProtoMessage() {}
func (m *Message) Reset() {
*m = Message{}
}
func (m *Message) String() string {
return string(m.Data)
}
func (m *Message) Marshal() ([]byte, error) {
return m.Data, nil
}
func (m *Message) Unmarshal(data []byte) error {
m.Data = data
return nil
}
func NewMessage(data []byte) *Message {
return &Message{data}
}

64
codec/proto/proto.go
View File

@@ -1,64 +0,0 @@
// Package proto provides a proto codec
package proto
import (
"io"
"io/ioutil"
"github.com/unistack-org/micro/v3/codec"
"google.golang.org/protobuf/proto"
)
type Codec struct {
Conn io.ReadWriteCloser
}
func (c *Codec) ReadHeader(m *codec.Message, t codec.MessageType) error {
return nil
}
func (c *Codec) ReadBody(b interface{}) error {
if b == nil {
return nil
}
buf, err := ioutil.ReadAll(c.Conn)
if err != nil {
return err
}
m, ok := b.(proto.Message)
if !ok {
return codec.ErrInvalidMessage
}
return proto.Unmarshal(buf, m)
}
func (c *Codec) Write(m *codec.Message, b interface{}) error {
if b == nil {
// Nothing to write
return nil
}
p, ok := b.(proto.Message)
if !ok {
return codec.ErrInvalidMessage
}
buf, err := proto.Marshal(p)
if err != nil {
return err
}
_, err = c.Conn.Write(buf)
return err
}
func (c *Codec) Close() error {
return c.Conn.Close()
}
func (c *Codec) String() string {
return "proto"
}
func NewCodec(c io.ReadWriteCloser) codec.Codec {
return &Codec{
Conn: c,
}
}

238
codec/protorpc/envelope.pb.go
View File

@@ -1,238 +0,0 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.25.0
// protoc v3.6.1
// source: codec/protorpc/envelope.proto
package protorpc
import (
proto "github.com/golang/protobuf/proto"
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
reflect "reflect"
sync "sync"
)
const (
// Verify that this generated code is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
// Verify that runtime/protoimpl is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)
// This is a compile-time assertion that a sufficiently up-to-date version
// of the legacy proto package is being used.
const _ = proto.ProtoPackageIsVersion4
type Request struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
ServiceMethod string `protobuf:"bytes,1,opt,name=service_method,json=serviceMethod,proto3" json:"service_method,omitempty"`
Seq uint64 `protobuf:"fixed64,2,opt,name=seq,proto3" json:"seq,omitempty"`
}
func (x *Request) Reset() {
*x = Request{}
if protoimpl.UnsafeEnabled {
mi := &file_codec_protorpc_envelope_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *Request) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*Request) ProtoMessage() {}
func (x *Request) ProtoReflect() protoreflect.Message {
mi := &file_codec_protorpc_envelope_proto_msgTypes[0]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use Request.ProtoReflect.Descriptor instead.
func (*Request) Descriptor() ([]byte, []int) {
return file_codec_protorpc_envelope_proto_rawDescGZIP(), []int{0}
}
func (x *Request) GetServiceMethod() string {
if x != nil {
return x.ServiceMethod
}
return ""
}
func (x *Request) GetSeq() uint64 {
if x != nil {
return x.Seq
}
return 0
}
type Response struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
ServiceMethod string `protobuf:"bytes,1,opt,name=service_method,json=serviceMethod,proto3" json:"service_method,omitempty"`
Seq uint64 `protobuf:"fixed64,2,opt,name=seq,proto3" json:"seq,omitempty"`
Error string `protobuf:"bytes,3,opt,name=error,proto3" json:"error,omitempty"`
}
func (x *Response) Reset() {
*x = Response{}
if protoimpl.UnsafeEnabled {
mi := &file_codec_protorpc_envelope_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *Response) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*Response) ProtoMessage() {}
func (x *Response) ProtoReflect() protoreflect.Message {
mi := &file_codec_protorpc_envelope_proto_msgTypes[1]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use Response.ProtoReflect.Descriptor instead.
func (*Response) Descriptor() ([]byte, []int) {
return file_codec_protorpc_envelope_proto_rawDescGZIP(), []int{1}
}
func (x *Response) GetServiceMethod() string {
if x != nil {
return x.ServiceMethod
}
return ""
}
func (x *Response) GetSeq() uint64 {
if x != nil {
return x.Seq
}
return 0
}
func (x *Response) GetError() string {
if x != nil {
return x.Error
}
return ""
}
var File_codec_protorpc_envelope_proto protoreflect.FileDescriptor
var file_codec_protorpc_envelope_proto_rawDesc = []byte{
0x0a, 0x1d, 0x63, 0x6f, 0x64, 0x65, 0x63, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x72, 0x70, 0x63,
0x2f, 0x65, 0x6e, 0x76, 0x65, 0x6c, 0x6f, 0x70, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12,
0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x72, 0x70, 0x63, 0x22, 0x42, 0x0a, 0x07, 0x52, 0x65, 0x71,
0x75, 0x65, 0x73, 0x74, 0x12, 0x25, 0x0a, 0x0e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x5f,
0x6d, 0x65, 0x74, 0x68, 0x6f, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d, 0x73, 0x65,
0x72, 0x76, 0x69, 0x63, 0x65, 0x4d, 0x65, 0x74, 0x68, 0x6f, 0x64, 0x12, 0x10, 0x0a, 0x03, 0x73,
0x65, 0x71, 0x18, 0x02, 0x20, 0x01, 0x28, 0x06, 0x52, 0x03, 0x73, 0x65, 0x71, 0x22, 0x59, 0x0a,
0x08, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x25, 0x0a, 0x0e, 0x73, 0x65, 0x72,
0x76, 0x69, 0x63, 0x65, 0x5f, 0x6d, 0x65, 0x74, 0x68, 0x6f, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28,
0x09, 0x52, 0x0d, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x4d, 0x65, 0x74, 0x68, 0x6f, 0x64,
0x12, 0x10, 0x0a, 0x03, 0x73, 0x65, 0x71, 0x18, 0x02, 0x20, 0x01, 0x28, 0x06, 0x52, 0x03, 0x73,
0x65, 0x71, 0x12, 0x14, 0x0a, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x18, 0x03, 0x20, 0x01, 0x28,
0x09, 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (
file_codec_protorpc_envelope_proto_rawDescOnce sync.Once
file_codec_protorpc_envelope_proto_rawDescData = file_codec_protorpc_envelope_proto_rawDesc
)
func file_codec_protorpc_envelope_proto_rawDescGZIP() []byte {
file_codec_protorpc_envelope_proto_rawDescOnce.Do(func() {
file_codec_protorpc_envelope_proto_rawDescData = protoimpl.X.CompressGZIP(file_codec_protorpc_envelope_proto_rawDescData)
})
return file_codec_protorpc_envelope_proto_rawDescData
}
var file_codec_protorpc_envelope_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
var file_codec_protorpc_envelope_proto_goTypes = []interface{}{
(*Request)(nil), // 0: protorpc.Request
(*Response)(nil), // 1: protorpc.Response
}
var file_codec_protorpc_envelope_proto_depIdxs = []int32{
0, // [0:0] is the sub-list for method output_type
0, // [0:0] is the sub-list for method input_type
0, // [0:0] is the sub-list for extension type_name
0, // [0:0] is the sub-list for extension extendee
0, // [0:0] is the sub-list for field type_name
}
func init() { file_codec_protorpc_envelope_proto_init() }
func file_codec_protorpc_envelope_proto_init() {
if File_codec_protorpc_envelope_proto != nil {
return
}
if !protoimpl.UnsafeEnabled {
file_codec_protorpc_envelope_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*Request); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
file_codec_protorpc_envelope_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*Response); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: file_codec_protorpc_envelope_proto_rawDesc,
NumEnums: 0,
NumMessages: 2,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_codec_protorpc_envelope_proto_goTypes,
DependencyIndexes: file_codec_protorpc_envelope_proto_depIdxs,
MessageInfos: file_codec_protorpc_envelope_proto_msgTypes,
}.Build()
File_codec_protorpc_envelope_proto = out.File
file_codec_protorpc_envelope_proto_rawDesc = nil
file_codec_protorpc_envelope_proto_goTypes = nil
file_codec_protorpc_envelope_proto_depIdxs = nil
}

21
codec/protorpc/envelope.pb.micro.go
View File

@@ -1,21 +0,0 @@
// Code generated by protoc-gen-micro. DO NOT EDIT.
// source: codec/protorpc/envelope.proto
package protorpc
import (
fmt "fmt"
proto "github.com/golang/protobuf/proto"
math "math"
)
// Reference imports to suppress errors if they are not otherwise used.
var _ = proto.Marshal
var _ = fmt.Errorf
var _ = math.Inf
// This is a compile-time assertion to ensure that this generated file
// is compatible with the proto package it is being compiled against.
// A compilation error at this line likely means your copy of the
// proto package needs to be updated.
const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package

14
codec/protorpc/envelope.proto
View File

@@ -1,14 +0,0 @@
syntax = "proto3";
package protorpc;
message Request {
string service_method = 1;
fixed64 seq = 2;
}
message Response {
string service_method = 1;
fixed64 seq = 2;
string error = 3;
}

36
codec/protorpc/netstring.go
View File

@@ -1,36 +0,0 @@
package protorpc
import (
"encoding/binary"
"io"
)
// WriteNetString writes data to a big-endian netstring on a Writer.
// Size is always a 32-bit unsigned int.
func WriteNetString(w io.Writer, data []byte) (written int, err error) {
size := make([]byte, 4)
binary.BigEndian.PutUint32(size, uint32(len(data)))
if written, err = w.Write(size); err != nil {
return
}
return w.Write(data)
}
// ReadNetString reads data from a big-endian netstring.
func ReadNetString(r io.Reader) (data []byte, err error) {
sizeBuf := make([]byte, 4)
_, err = r.Read(sizeBuf)
if err != nil {
return nil, err
}
size := binary.BigEndian.Uint32(sizeBuf)
if size == 0 {
return nil, nil
}
data = make([]byte, size)
_, err = r.Read(data)
if err != nil {
return nil, err
}
return
}

186
codec/protorpc/protorpc.go
View File

@@ -1,186 +0,0 @@
// Protorpc provides a net/rpc proto-rpc codec. See envelope.proto for the format.
package protorpc
import (
"bytes"
"fmt"
"io"
"strconv"
"sync"
"github.com/golang/protobuf/proto"
"github.com/unistack-org/micro/v3/codec"
)
type flusher interface {
Flush() error
}
type protoCodec struct {
sync.Mutex
rwc io.ReadWriteCloser
mt codec.MessageType
buf *bytes.Buffer
}
func (c *protoCodec) Close() error {
c.buf.Reset()
return c.rwc.Close()
}
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: id(m.Id)}
data, err := proto.Marshal(pbr)
if err != nil {
return err
}
_, err = WriteNetString(c.rwc, data)
if err != nil {
return err
}
// dont trust or incoming message
m, ok := b.(proto.Message)
if !ok {
return codec.ErrInvalidMessage
}
data, err = proto.Marshal(m)
if err != nil {
return err
}
_, err = WriteNetString(c.rwc, data)
if err != nil {
return err
}
if flusher, ok := c.rwc.(flusher); ok {
if err = flusher.Flush(); err != nil {
return err
}
}
case codec.Response, codec.Error:
c.Lock()
defer c.Unlock()
rtmp := &Response{ServiceMethod: m.Method, Seq: id(m.Id), Error: m.Error}
data, err := proto.Marshal(rtmp)
if err != nil {
return err
}
_, err = WriteNetString(c.rwc, data)
if err != nil {
return err
}
if pb, ok := b.(proto.Message); ok {
data, err = proto.Marshal(pb)
if err != nil {
return err
}
} else {
data = nil
}
_, err = WriteNetString(c.rwc, data)
if err != nil {
return err
}
if flusher, ok := c.rwc.(flusher); ok {
if err = flusher.Flush(); err != nil {
return err
}
}
case codec.Event:
m, ok := b.(proto.Message)
if !ok {
return codec.ErrInvalidMessage
}
data, err := proto.Marshal(m)
if err != nil {
return err
}
c.rwc.Write(data)
default:
return fmt.Errorf("Unrecognised message type: %v", m.Type)
}
return nil
}
func (c *protoCodec) ReadHeader(m *codec.Message, mt codec.MessageType) error {
c.buf.Reset()
c.mt = mt
switch mt {
case codec.Request:
data, err := ReadNetString(c.rwc)
if err != nil {
return err
}
rtmp := new(Request)
err = proto.Unmarshal(data, rtmp)
if err != nil {
return err
}
m.Method = rtmp.GetServiceMethod()
m.Id = fmt.Sprintf("%d", rtmp.GetSeq())
case codec.Response:
data, err := ReadNetString(c.rwc)
if err != nil {
return err
}
rtmp := new(Response)
err = proto.Unmarshal(data, rtmp)
if err != nil {
return err
}
m.Method = rtmp.GetServiceMethod()
m.Id = fmt.Sprintf("%d", rtmp.GetSeq())
m.Error = rtmp.GetError()
case codec.Event:
_, err := io.Copy(c.buf, c.rwc)
return err
default:
return fmt.Errorf("Unrecognised message type: %v", mt)
}
return nil
}
func (c *protoCodec) ReadBody(b interface{}) error {
var data []byte
switch c.mt {
case codec.Request, codec.Response:
var err error
data, err = ReadNetString(c.rwc)
if err != nil {
return err
}
case codec.Event:
data = c.buf.Bytes()
default:
return fmt.Errorf("Unrecognised message type: %v", c.mt)
}
if b != nil {
return proto.Unmarshal(data, b.(proto.Message))
}
return nil
}
func NewCodec(rwc io.ReadWriteCloser) codec.Codec {
return &protoCodec{
buf: bytes.NewBuffer(nil),
rwc: rwc,
}
}

80
codec/text/text.go
View File

@@ -1,80 +0,0 @@
// Package text reads any text/* content-type
package text
import (
"fmt"
"io"
"io/ioutil"
"github.com/unistack-org/micro/v3/codec"
)
type Codec struct {
Conn io.ReadWriteCloser
}
// Frame gives us the ability to define raw data to send over the pipes
type Frame struct {
Data []byte
}
func (c *Codec) ReadHeader(m *codec.Message, t codec.MessageType) error {
return nil
}
func (c *Codec) ReadBody(b interface{}) error {
// read bytes
buf, err := ioutil.ReadAll(c.Conn)
if err != nil {
return err
}
switch v := b.(type) {
case *string:
*v = string(buf)
case *[]byte:
*v = buf
case *Frame:
v.Data = buf
default:
return fmt.Errorf("failed to read body: %v is not type of *[]byte", b)
}
return nil
}
func (c *Codec) Write(m *codec.Message, b interface{}) error {
var v []byte
switch ve := b.(type) {
case nil:
return nil
case *Frame:
v = ve.Data
case *[]byte:
v = *ve
case *string:
v = []byte(*ve)
case string:
v = []byte(ve)
case []byte:
v = ve
default:
return fmt.Errorf("failed to write: %v is not type of *[]byte or []byte", b)
}
_, err := c.Conn.Write(v)
return err
}
func (c *Codec) Close() error {
return c.Conn.Close()
}
func (c *Codec) String() string {
return "text"
}
func NewCodec(c io.ReadWriteCloser) codec.Codec {
return &Codec{
Conn: c,
}
}