Improved structure added test

client/wsrpc.js

@@ -43,6 +43,8 @@
       switch (obj.op) {
         case 'call':
           if (obj.method === undefined || this._exports[obj.method] === undefined) {
+            var dataObj = { 'op': 'response', 'id': obj.id, 'error': 'Method not found' }
+            ctx.ws.send(JSON.stringify(dataObj))
             // TODO: Send object to inform error
             return
           }
@@ -58,7 +60,7 @@
           this._exports[obj.method].apply(this._exports[obj.method], nparams)
           break
         case 'response':
-          ctx._requests[obj.id](obj.response)
+          ctx._requests[obj.id](obj.response, obj.error)
           delete ctx._requests[obj.reqId]
           break
       }

clientctx.go

@@ -2,6 +2,7 @@
 package wsrpc
 
 import (
+	"context"
 	"errors"
 	"log"
 	"reflect"
@@ -12,6 +13,7 @@ import (
 	"golang.org/x/net/websocket"
 )
 
+// Vars
 var (
 	ErrNParam            = errors.New("Invalid number of parameters")
 	ErrNReturn           = errors.New("Number of outputs mismatch")
@@ -20,14 +22,17 @@ var (
 	errorInterface = reflect.TypeOf((*error)(nil)).Elem()
 )
 
-// Call object
+// CallObj object
+// Message
 type CallObj struct {
 	OP     string `json:"op"`     // operation
 	ID     string `json:"id"`     // Id of the call
 	Method string `json:"method"` // could be param 0 instead?
 
+	// Payload
 	Params   []interface{} `json:"params"`
 	Response interface{}   `json:"response"`
+	Error    string        `json:"error"`
 }
 
 // DataObj common structure for dymanic json object
@@ -40,61 +45,87 @@ type ListenerFunc func(...interface{}) (interface{}, error)
 
 // ClientCtx main client struct
 type ClientCtx struct {
-	locker sync.Mutex
+	context.Context
+	Close func()
 
-	WS *websocket.Conn
+	locker sync.Mutex
+	WS     *websocket.Conn
 	// stuff
 	listeners map[string]ListenerFunc
-	requests  map[string]chan interface{}
+	requests  map[string]chan CallObj
+	message   chan CallObj
 }
 
-//NewHandler creates a new WsRPC client handler
-func NewHandler(id string, ws *websocket.Conn) *ClientCtx {
-	var c = ClientCtx{
+// NewClient new client (should have background context here)
+func NewClient(ws *websocket.Conn) *ClientCtx {
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	return &ClientCtx{
+		Context:   ctx,
+		Close:     cancelFunc,
 		WS:        ws,
 		listeners: map[string]ListenerFunc{},
-		requests:  map[string]chan interface{}{},
+		requests:  map[string]chan CallObj{},
 	}
-	return &c
+
 }
 
-// Process messages
-func (c *ClientCtx) Process(data CallObj) {
-	switch data.OP {
-	case "call":
-		params := data.Params
-		var idn = data.Method
-		var reqID = data.ID
-		var fn, ok = c.listeners[idn]
-		if !ok {
+// Process received messages
+func (c *ClientCtx) Process() {
+	for {
+		select {
+		case <-c.Done():
+			return // Break loop
+		default:
+		}
+		var data = CallObj{}
+		err := websocket.JSON.Receive(c.WS, &data)
+		if err != nil { // Close receive
+			c.Close()
 			return
 		}
-		go func() { // async send
-			ret, err := fn(params...)
-			if err != nil {
-				log.Println("Create error response, panic?")
+
+		switch data.OP {
+		case "call":
+			params := data.Params
+			var idn = data.Method
+			var reqID = data.ID
+			var fn, ok = c.listeners[idn]
+			if !ok {
+				websocket.JSON.Send(c.WS, CallObj{
+					OP:    "response",
+					ID:    reqID,
+					Error: "Method not found",
+				})
+				break
 			}
-			var responseObj = CallObj{
-				OP:       "response",
-				ID:       reqID,
-				Response: ret,
+			go func() { // async send
+				errStr := ""
+				ret, err := fn(params...)
+				if err != nil {
+					errStr = err.Error()
+				}
+				var responseObj = CallObj{
+					OP:       "response",
+					ID:       reqID,
+					Response: ret,
+					Error:    errStr,
+				}
+				websocket.JSON.Send(c.WS, responseObj)
+			}()
+		case "response":
+			c.locker.Lock()
+			mchan, ok := c.requests[data.ID]
+			delete(c.requests, data.ID)
+			c.locker.Unlock()
+			if ok {
+				mchan <- data
 			}
-			//log.Println("Sending response")
-			websocket.JSON.Send(c.WS, responseObj)
-		}()
-	case "response":
-		c.locker.Lock()
-		mchan, ok := c.requests[data.ID]
-		delete(c.requests, data.ID)
-		c.locker.Unlock()
-		if ok {
-			mchan <- data.Response
 		}
 	}
 }
 
 // Call a client method and estabilishes a request id
-func (c *ClientCtx) Call(method string, params ...interface{}) interface{} {
+func (c *ClientCtx) Call(method string, params ...interface{}) (interface{}, error) {
 	u := uuid.New()
 	uuidStr := u.String()
 
@@ -105,20 +136,33 @@ func (c *ClientCtx) Call(method string, params ...interface{}) interface{} {
 		Params: params,
 	}
 
-	res := make(chan interface{}, 1)
+	resCh := make(chan CallObj, 1)
+	// Store the channel
 	c.locker.Lock()
-	c.requests[uuidStr] = res
+	c.requests[uuidStr] = resCh
 	c.locker.Unlock()
+
+	// IO send
+	// Send to dispatcher instead?
 	websocket.JSON.Send(c.WS, &callObj)
 
-	return <-res // Block until value
+	// Hang until response
+	res := <-resCh // Wait for response
+
+	// Got Response
+	var err error
+	if res.Error != "" {
+		err = errors.New(res.Error)
+	}
+	return res.Response, err
 }
 
-//On add a event listener on browser
+//Define export a method to ws client
 func (c *ClientCtx) Define(name string, listener ListenerFunc) {
 	c.listeners[name] = listener
 }
 
+//Export a struct of funcs
 func (c *ClientCtx) Export(obj interface{}) {
 	// Reflect here
 	typ := reflect.TypeOf(obj)

wsrpc.go

@@ -3,7 +3,6 @@ package wsrpc
 //go:generate folder2go client wsrpcAssets
 
 import (
-	"log"
 	"net/http"
 	"strings"
 
@@ -55,31 +54,10 @@ type Func func(*ClientCtx)
 
 // Create a client websocket handler on connection
 func createCliHandler(handler Func) websocket.Handler {
+	// Move this to clientCTX handler
 	return func(ws *websocket.Conn) {
-		ch := NewHandler("main", ws)
-		go handler(ch)
-		for {
-			var data = CallObj{}
-			err := websocket.JSON.Receive(ws, &data)
-			if err != nil {
-				log.Println("Error: ", err)
-				return
-			}
-			ch.Process(data)
-		}
+		cli := NewClient(ws)
+		go handler(cli) // Paralell launch a client initiator
+		cli.Process()   // blocks
 	}
 }
-
-// RegisterTo register both js and websocket in the servermux
-/*func RegisterTo(server ServerContext, handler HandleFunc) {
-	server.Handle("/wsrpc", websocket.Handler(createCliHandler(handler)))
-	server.HandleFunc("/wsrpc/client.js", wsrpcjsFunc)
-}
-
-// Handler we can attatch this to http.Server
-func Handler(handler HandleFunc) http.Handler {
-	mux := http.NewServeMux()
-	RegisterTo(mux, handler)
-	// When we receive a connection
-	return mux
-}*/

wsrpc_test.go

@@ -0,0 +1,126 @@
+package wsrpc_test
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+	"time"
+
+	"github.com/gohxs/prettylog"
+	"golang.org/x/net/websocket"
+
+	"dev.hexasoftware.com/stdio/wsrpc"
+)
+
+func init() {
+	prettylog.Global()
+}
+
+var (
+	ready = make(chan struct{})
+)
+
+///////////////////////////
+// TESTS
+//////////////
+
+func TestCall(t *testing.T) {
+
+	ws := prepareClient(t) // Prepare a temp server and a client
+
+	// Sending
+	callObj := wsrpc.CallObj{
+		OP:     "call",
+		ID:     "123",
+		Method: "hello",
+		Params: []interface{}{1, 2},
+	}
+	websocket.JSON.Send(ws, callObj)
+
+	//Receive
+	res := wsrpc.CallObj{}
+	err := websocket.JSON.Receive(ws, &res)
+	if err != nil {
+		t.Fatal("Read error", err)
+	}
+	if res.OP != "response" || res.Response != "Hello world" {
+		t.Fatal("Failed, invalid response")
+	}
+	if res.Error != "" {
+		t.Fatal("Remote error", res.Error)
+	}
+
+}
+func TestParams(t *testing.T) {
+	ws := prepareClient(t)
+	cli := wsrpc.NewClient(ws)
+
+	go cli.Process()
+	res, err := cli.Call("sum", 1, 2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res != float64(3) {
+		t.Fatal("Message error")
+	}
+}
+
+func TestError(t *testing.T) {
+	ws := prepareClient(t)
+
+	cli := wsrpc.NewClient(ws)
+
+	go cli.Process()
+	_, err := cli.Call("nomethod", 1, 2, 3, 4)
+	if err == nil {
+		t.Fatal("It should return an error but didn't")
+	}
+	_, err = cli.Call("hello", 1, 2, 3)
+	if err != nil {
+		t.Fatal("Ups")
+	}
+
+}
+
+//////////////////////
+// HELPERS
+////////
+func cliTest(ws *wsrpc.ClientCtx) {
+
+	ws.Define("hello", func(params ...interface{}) (interface{}, error) {
+		return "Hello world", nil
+	})
+	ws.Define("done", func(params ...interface{}) (interface{}, error) {
+		return nil, nil
+	})
+	ws.Define("sum", func(params ...interface{}) (interface{}, error) {
+		return params[0].(float64) + params[1].(float64), nil
+	})
+
+	ready <- struct{}{}
+
+	<-time.After(10 * time.Second)
+	ws.WS.Close()
+}
+
+func prepareClient(t *testing.T) *websocket.Conn {
+
+	mux := http.NewServeMux()
+	mux.Handle("/", wsrpc.New(cliTest))
+
+	serv := httptest.NewServer(mux)
+
+	u, err := url.Parse(serv.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	wsURL := fmt.Sprintf("ws://%s:%s/wsrpc", u.Hostname(), u.Port())
+	ws, err := websocket.Dial(wsURL, "", "http://localhost/")
+	if err != nil {
+		t.Fatal(err)
+	}
+	<-ready // wait until define ready
+	return ws
+}