init

.coveralls.yml

@@ -0,0 +1,3 @@
+service_name: travis-ci
+coverage_clover: clover.xml
+json_path: clover.json

.gitattributes

@@ -0,0 +1 @@
+/tests export-ignore

.gitignore

@@ -0,0 +1,6 @@
+.idea/
+composer.lock
+vendor/
+clover.xml
+coveralls.json
+tests/*.exe

README.md

@@ -0,0 +1 @@
+WIP

composer.json

@@ -0,0 +1,30 @@
+{
+  "name": "spiral/goridge",
+  "type": "goridge",
+  "description": "Dead simple, high performance, drop-in bridge to Golang RPC with zero dependencies",
+  "license": "MIT",
+  "authors": [
+    {
+      "name": "Anton Titov / Wolfy-J",
+      "email": "wolfy.jd@gmail.com"
+    }
+  ],
+  "require": {
+    "php": ">=7.1"
+  },
+  "require-dev": {
+    "phpunit/phpunit": "~6.0",
+    "mockery/mockery": "^0.9.9",
+    "symfony/process": "^3.3"
+  },
+  "autoload": {
+    "psr-4": {
+      "Spiral\\": "source/"
+    }
+  },
+  "autoload-dev": {
+    "psr-4": {
+      "Spiral\\Tests\\": "tests/Cases/"
+    }
+  }
+}

flags.go

@@ -0,0 +1,11 @@
+// Author Wolfy-J, 2017. License MIT
+
+package goridge
+
+const (
+	KeepConnection  = 0
+	CloseConnection = 1
+	NoBody          = 16
+	RawBody         = 32
+	ErrorBody       = 64
+)

goridge.go

@@ -0,0 +1,156 @@
+// Author Wolfy-J, 2017. License MIT
+
+package goridge
+
+import (
+	"net/rpc"
+	"io"
+	"encoding/json"
+	"sync"
+	"log"
+	"errors"
+	"reflect"
+)
+
+const (
+	ChunkSize = 655336
+)
+
+// A JSONCodec implemented ServeCodec interface using 9 bytes data prefixes
+// and data marshaling via json. This is buffering like codec.
+type JSONCodec struct {
+	mu     sync.Mutex // concurrent write
+	rwc    io.ReadWriteCloser
+	prefix Prefix // next package prefix
+
+	closed bool
+}
+
+// NewJSONCodec initiates new JSONCodec over given connection.
+func NewJSONCodec(conn io.ReadWriteCloser) *JSONCodec {
+	return &JSONCodec{
+		rwc:    conn,
+		prefix: Prefix(make([]byte, 9)),
+	}
+}
+
+// ReadRequestHeader expects to read prefixed method name to be called.
+func (c *JSONCodec) ReadRequestHeader(r *rpc.Request) error {
+	if _, err := c.rwc.Read(c.prefix); err != nil {
+		return err
+	}
+
+	if !c.prefix.HasBody() {
+		return nil
+	}
+
+	method := make([]byte, c.prefix.Size())
+	if _, err := c.rwc.Read(method); err != nil {
+		return err
+	}
+
+	r.ServiceMethod = string(method)
+	return nil
+}
+
+// ReadRequestBody fetches prefixed body payload and automatically unmarshal it as json is no RawBody flag are set.
+func (c *JSONCodec) ReadRequestBody(out interface{}) error {
+	if _, err := c.rwc.Read(c.prefix); err != nil {
+		return err
+	}
+
+	if !c.prefix.HasBody() {
+		return nil
+	}
+
+	// more efficient vs more memory?
+	body := make([]byte, c.prefix.Size())
+	body = body[:0]
+
+	buffer := make([]byte, min(uint64(cap(body)), ChunkSize))
+	doneBytes := uint64(0)
+
+	// read only prefix.Size() from socket
+	for {
+		n, err := c.rwc.Read(buffer)
+		if err != nil {
+			return err
+		}
+
+		body = append(body, buffer[:n]...)
+		doneBytes += uint64(n)
+
+		if doneBytes == c.prefix.Size() {
+			break
+		}
+	}
+
+	if c.prefix.Flags()&RawBody == RawBody {
+		if bin, ok := out.(*[]byte); ok {
+			*bin = append(*bin, body...)
+			return nil
+		} else {
+			return errors.New("{rawData} request for " + reflect.ValueOf(out).Elem().Kind().String())
+		}
+
+		return nil
+	}
+
+	return json.Unmarshal(body, out)
+}
+
+// WriteResponse marshaled response, byte slice or error to remote party.
+func (c *JSONCodec) WriteResponse(r *rpc.Response, body interface{}) error {
+	if r.Error != "" {
+		log.Println("rpc: goridge error response:", r.Error)
+		return c.write([]byte(r.Error), CloseConnection|ErrorBody|RawBody)
+	}
+
+	if bin, ok := body.(*[]byte); ok {
+		return c.write(*bin, KeepConnection|RawBody)
+	}
+
+	res, err := json.Marshal(body)
+	if err != nil {
+		return err
+	}
+
+	return c.write(res, KeepConnection)
+}
+
+// Close underlying socket.
+func (c *JSONCodec) Close() error {
+	if c.closed {
+		return nil
+	}
+
+	c.closed = true
+	return c.rwc.Close()
+}
+
+// Write data prefix [flag][size uint64 LE] and payload into socket.
+func (c *JSONCodec) write(payload []byte, flags byte) error {
+	prefix := Prefix(make([]byte, 9))
+	prefix.SetFlags(flags)
+	prefix.SetSize(uint64(len(payload)))
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if _, err := c.rwc.Write(prefix); err != nil {
+		return err
+	}
+
+	if _, err := c.rwc.Write(payload); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func min(a, b uint64) uint64 {
+	if a < b {
+		return a
+	}
+	return b
+}

prefix.go

@@ -0,0 +1,30 @@
+// Author Wolfy-J, 2017. License MIT
+
+package goridge
+
+import (
+	"encoding/binary"
+)
+
+// Prefix is always 9 bytes long and contain meta flags and length of next data package.
+type Prefix []byte
+
+func (p Prefix) Flags() byte {
+	return p[0]
+}
+
+func (p Prefix) SetFlags(flags byte) {
+	p[0] = p[0] | flags
+}
+
+func (p Prefix) Size() uint64 {
+	return binary.LittleEndian.Uint64(p[1:])
+}
+
+func (p Prefix) HasBody() bool {
+	return p.Flags()&NoBody == 0 && p.Size() != 0
+}
+
+func (p Prefix) SetSize(size uint64) {
+	binary.LittleEndian.PutUint64(p[1:], size)
+}