wspr: Treat WSPR as an RPC service, part 5.
Transitioning namespace calls to an RPC interface.
MultiPart: 2/2
Change-Id: Ice8a5b90258b3e8eb893c2b8aebd263ebb6920dc
diff --git a/services/wsprd/app/app.go b/services/wsprd/app/app.go
index 32231b8..2e959e0 100644
--- a/services/wsprd/app/app.go
+++ b/services/wsprd/app/app.go
@@ -103,6 +103,11 @@
// Store for all the Blessings that javascript has a handle to.
blessingsStore *principal.JSBlessingsHandles
+
+ // reservedServices contains a map of reserved service names. These
+ // are objects that serve requests in wspr without actually making
+ // an outgoing rpc call.
+ reservedServices map[string]ipc.Invoker
}
// NewController creates a new Controller. writerCreator will be used to create a new flow for rpcs to
@@ -129,6 +134,19 @@
blessingsStore: principal.NewJSBlessingsHandles(),
}
+ controllerInvoker, err := ipc.ReflectInvoker(ControllerServer(controller))
+ if err != nil {
+ return nil, err
+ }
+ namespaceInvoker, err := ipc.ReflectInvoker(namespace.New(ctx))
+ if err != nil {
+ return nil, err
+ }
+ controller.reservedServices = map[string]ipc.Invoker{
+ "__controller": controllerInvoker,
+ "__namespace": namespaceInvoker,
+ }
+
controller.setup()
return controller, nil
}
@@ -347,9 +365,23 @@
ctx *context.T
vrpc *VeyronRPCRequest
tags []*vdl.Value
+ w lib.ClientWriter
}
-func (l *localCall) Send(interface{}) error { return nil }
+func (l *localCall) Send(item interface{}) error {
+ vomItem, err := lib.VomEncode(item)
+ if err != nil {
+ err = verror.New(marshallingError, l.ctx, item, err)
+ l.w.Error(err)
+ return err
+ }
+ if err := l.w.Send(lib.ResponseStream, vomItem); err != nil {
+ err = verror.New(marshallingError, l.ctx, item)
+ l.w.Error(err)
+ return err
+ }
+ return nil
+}
func (l *localCall) Recv(interface{}) error { return nil }
func (l *localCall) Blessings() security.Blessings { return nil }
func (l *localCall) Server() ipc.Server { return nil }
@@ -366,12 +398,7 @@
func (l *localCall) LocalEndpoint() naming.Endpoint { return nil }
func (l *localCall) RemoteEndpoint() naming.Endpoint { return nil }
-func (c *Controller) handleInternalCall(ctx *context.T, msg *VeyronRPCRequest, decoder *vom.Decoder, w lib.ClientWriter, span vtrace.Span) {
- invoker, err := ipc.ReflectInvoker(ControllerServer(c))
- if err != nil {
- w.Error(verror.Convert(verror.ErrInternal, ctx, err))
- return
- }
+func (c *Controller) handleInternalCall(ctx *context.T, invoker ipc.Invoker, msg *VeyronRPCRequest, decoder *vom.Decoder, w lib.ClientWriter, span vtrace.Span) {
argptrs, tags, err := invoker.Prepare(msg.Method, int(msg.NumInArgs))
if err != nil {
w.Error(verror.Convert(verror.ErrInternal, ctx, err))
@@ -383,11 +410,17 @@
return
}
}
- results, err := invoker.Invoke(msg.Method, &localCall{ctx, msg, tags}, argptrs)
+ results, err := invoker.Invoke(msg.Method, &localCall{ctx, msg, tags, w}, argptrs)
if err != nil {
w.Error(verror.Convert(verror.ErrInternal, ctx, err))
return
}
+ if msg.IsStreaming {
+ if err := w.Send(lib.ResponseStreamClose, nil); err != nil {
+ w.Error(verror.New(marshallingError, ctx, "ResponseStreamClose"))
+ }
+ }
+
// Convert results from []interface{} to []*vdl.Value.
vresults := make([]*vdl.Value, len(results))
for i, res := range results {
@@ -436,8 +469,8 @@
}
// If this message is for an internal service, do a short-circuit dispatch here.
- if msg.Name == "controller" {
- go c.handleInternalCall(ctx, &msg, decoder, w, span)
+ if invoker, ok := c.reservedServices[msg.Name]; ok {
+ go c.handleInternalCall(ctx, invoker, &msg, decoder, w, span)
return
}
@@ -701,8 +734,3 @@
blessings, _ := call.RemoteBlessings()
return blessings, nil
}
-
-// HandleNamespaceRequest uses the namespace client to respond to namespace specific requests such as glob
-func (c *Controller) HandleNamespaceRequest(ctx *context.T, data string, w lib.ClientWriter) {
- namespace.HandleRequest(ctx, data, w)
-}
diff --git a/services/wsprd/app/app_test.go b/services/wsprd/app/app_test.go
index 7a334ce..7fd7106 100644
--- a/services/wsprd/app/app_test.go
+++ b/services/wsprd/app/app_test.go
@@ -363,7 +363,7 @@
veyron2.GetNamespace(controller.Context()).SetRoots("/" + endpoint.String())
req, err := makeRequest(VeyronRPCRequest{
- Name: "controller",
+ Name: "__controller",
Method: "Serve",
NumInArgs: 2,
NumOutArgs: 1,
diff --git a/services/wsprd/app/messaging.go b/services/wsprd/app/messaging.go
index a737889..7865870 100644
--- a/services/wsprd/app/messaging.go
+++ b/services/wsprd/app/messaging.go
@@ -116,9 +116,6 @@
case AuthResponseMessage:
go c.HandleAuthResponse(msg.Id, msg.Data)
- case NamespaceRequestMessage:
- go c.HandleNamespaceRequest(ctx, msg.Data, w)
-
default:
w.Error(verror.New(errUnknownMessageType, ctx, msg.Type))
}
