diff --git a/backend/app/store/remote/client.go b/backend/app/store/remote/client.go new file mode 100644 index 00000000..1fb8ed2f --- /dev/null +++ b/backend/app/store/remote/client.go @@ -0,0 +1,53 @@ +package remote + +import ( + "bytes" + "encoding/json" + "net/http" + + "github.com/pkg/errors" +) + +// Client implements remote engine and delegates all calls to remote http server +type Client struct { + API string + Client http.Client + AuthUser string + AuthPasswd string +} + +// Call remote server with given method and arguments +func (r *Client) Call(method string, args ...interface{}) (*Response, error) { + + b, err := json.Marshal(Request{Method: method, Params: args}) + if err != nil { + return nil, errors.Wrapf(err, "marshaling failed for %s", method) + } + + req, err := http.NewRequest("POST", r.API, bytes.NewBuffer(b)) + if err != nil { + return nil, errors.Wrapf(err, "failed to make request for %s", method) + } + + if r.AuthUser != "" && r.AuthPasswd != "" { + req.SetBasicAuth(r.AuthUser, r.AuthPasswd) + } + resp, err := r.Client.Do(req) + if err != nil { + return nil, errors.Wrapf(err, "remote call failed for %s", method) + } + defer resp.Body.Close() + if resp.StatusCode != 200 { + return nil, errors.Errorf("bad status %d for %s", resp.StatusCode, method) + } + + cr := Response{} + if err = json.NewDecoder(resp.Body).Decode(&cr); err != nil { + return nil, errors.Wrapf(err, "failed to decode response for %s", method) + } + + if cr.Error != "" { + return nil, errors.New(cr.Error) + } + return &cr, nil +} diff --git a/backend/app/store/remote/remote_test.go b/backend/app/store/remote/client_test.go similarity index 100% rename from backend/app/store/remote/remote_test.go rename to backend/app/store/remote/client_test.go diff --git a/backend/app/store/remote/remote.go b/backend/app/store/remote/remote.go index 77c68898..e72a41a9 100644 --- a/backend/app/store/remote/remote.go +++ b/backend/app/store/remote/remote.go @@ -1,20 +1,13 @@ +// Package remote implements client ans server for RPC-like communication with remote storage. +// The protocol is somewhat simplified version of json-rpc with a single POST call sending +// Request json (method name and the list of parameters) and receiving back json Response with "result" json +// and error string package remote import ( - "bytes" "encoding/json" - "net/http" - - "github.com/pkg/errors" ) -// Client implements remote engine and delegates all calls to remote http server -type Client struct { - API string - Client http.Client - AuthUser string - AuthPasswd string -} // Request encloses method name and all params type Request struct { @@ -27,37 +20,3 @@ type Response struct { Result *json.RawMessage `json:"result,omitempty"` Error string `json:"error,omitempty"` } - -// Call remote server with given method and arguments -func (r *Client) Call(method string, args ...interface{}) (*Response, error) { - - b, err := json.Marshal(Request{Method: method, Params: args}) - if err != nil { - return nil, errors.Wrapf(err, "marshaling failed for %s", method) - } - - req, err := http.NewRequest("POST", r.API, bytes.NewBuffer(b)) - if err != nil { - return nil, errors.Wrapf(err, "failed to make request for %s", method) - } - - req.SetBasicAuth(r.AuthUser, r.AuthPasswd) - resp, err := r.Client.Do(req) - if err != nil { - return nil, errors.Wrapf(err, "remote Call failed for %s", method) - } - defer resp.Body.Close() - if resp.StatusCode != 200 { - return nil, errors.Errorf("bad status %d for %s", resp.StatusCode, method) - } - - cr := Response{} - if err = json.NewDecoder(resp.Body).Decode(&cr); err != nil { - return nil, errors.Wrapf(err, "failed to decode response for %s", method) - } - - if cr.Error != "" { - return nil, errors.New(cr.Error) - } - return &cr, nil -} diff --git a/backend/app/store/remote/server.go b/backend/app/store/remote/server.go new file mode 100644 index 00000000..bfabef2a --- /dev/null +++ b/backend/app/store/remote/server.go @@ -0,0 +1,101 @@ +package remote + +import ( + "context" + "encoding/json" + "fmt" + "log" + "net/http" + "sync" + "time" + + "github.com/go-chi/chi" + "github.com/go-chi/render" + "github.com/pkg/errors" +) + +type Server struct { + CommandURL string + AuthUser string + AuthPasswd string + + funcs struct { + m map[string]ServerFn + once sync.Once + } + + httpServer struct { + *http.Server + sync.Mutex + } +} + +type ServerFn func(params *json.RawMessage) Response + +func (s *Server) Run(port int) error { + if s.funcs.m == nil && len(s.funcs.m) == 0 { + return errors.Errorf("nothing mapped for dispatch, Add has to be called prior to Run") + } + + router := chi.NewRouter() + + type request struct { + Method string `json:"method"` + Params *json.RawMessage `json:"params"` + } + + router.Post(s.CommandURL, func(w http.ResponseWriter, r *http.Request) { + req := request{} + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + fn, ok := s.funcs.m[req.Method] + if !ok { + w.WriteHeader(http.StatusNotImplemented) + return + } + render.JSON(w, r, fn(req.Params)) + }) + + s.httpServer.Lock() + s.httpServer.Server = &http.Server{ + Addr: fmt.Sprintf(":%d", port), + Handler: router, + ReadHeaderTimeout: 5 * time.Second, + WriteTimeout: 10 * time.Second, + IdleTimeout: 30 * time.Second, + } + s.httpServer.Unlock() + + log.Printf("[INFO] listen on %d", port) + return s.httpServer.ListenAndServe() +} + +func (s *Server) EncodeResponse(resp interface{}) (Response, error) { + v, err := json.Marshal(&resp) + if err != nil { + return Response{}, err + } + raw := json.RawMessage{} + if err = raw.UnmarshalJSON(v); err != nil { + return Response{}, err + } + return Response{Result: &raw}, nil +} + +func (s *Server) Shutdown() error { + s.httpServer.Lock() + defer s.httpServer.Unlock() + if s.httpServer.Server == nil { + return errors.Errorf("http server is not running") + } + return s.httpServer.Shutdown(context.TODO()) +} + +func (s *Server) Add(method string, fn ServerFn) { + s.funcs.once.Do(func() { + s.funcs.m = map[string]ServerFn{} + }) + s.funcs.m[method] = fn +} diff --git a/backend/app/store/remote/server_test.go b/backend/app/store/remote/server_test.go new file mode 100644 index 00000000..4d6a11dc --- /dev/null +++ b/backend/app/store/remote/server_test.go @@ -0,0 +1,67 @@ +package remote + +import ( + "bytes" + "encoding/json" + "io/ioutil" + "net/http" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestServer(t *testing.T) { + s := Server{CommandURL: "/v1/cmd"} + + type respData struct { + Res1 string + Res2 bool + } + + s.Add("test", func(params *json.RawMessage) Response { + args := []interface{}{} + if err := json.Unmarshal(*params, &args); err != nil { + return Response{Error: err.Error()} + } + t.Logf("%+v", args) + + assert.Equal(t, 4, len(args)) + assert.Equal(t, "blah", args[0].(string)) + assert.Equal(t, 42., args[1].(float64)) + assert.Equal(t, true, args[2].(bool)) + assert.Equal(t, "", args[3].(time.Time)) + + r, err := s.EncodeResponse(respData{"res blah", true}) + assert.NoError(t, err) + return r + }) + + go func() { s.Run(9091) }() + time.Sleep(10 * time.Millisecond) + + // check with direct http call + clientReq := Request{Method: "test", Params: []interface{}{"blah", 42, true, time.Date(2018, 6, 9, 16, 7, 25, 0, time.UTC)}} + b := bytes.Buffer{} + require.NoError(t, json.NewEncoder(&b).Encode(clientReq)) + resp, err := http.Post("http://127.0.0.1:9091/v1/cmd", "application/json", &b) + require.NoError(t, err) + defer resp.Body.Close() + assert.Equal(t, 200, resp.StatusCode) + data, err := ioutil.ReadAll(resp.Body) + assert.NoError(t, err) + assert.Equal(t, `{"result":{"Res1":"res blah","Res2":true}}`+"\n", string(data)) + + // check with client call + c := Client{API: "http://127.0.0.1:9091/v1/cmd", Client: http.Client{}} + r, err := c.Call("test", "blah", 42, true, time.Date(2018, 6, 9, 16, 7, 25, 0, time.UTC)) + assert.NoError(t, err) + assert.Equal(t, "", r.Error) + + res := respData{} + err = json.Unmarshal(*r.Result, &res) + assert.Equal(t, respData{Res1: "res blah", Res2: true}, res) + + assert.NoError(t, s.Shutdown()) +}