services/xproxyd: Make proxy use now public flow message.
Rather than vom encoding route information we know actually read
the setup message from the flow and retrieve the route information
from there.
Change-Id: Ia52502f3608731be6f136c7ff36e25bfb3405b98
diff --git a/runtime/internal/flow/manager/manager.go b/runtime/internal/flow/manager/manager.go
index 748a5bc..122264a 100644
--- a/runtime/internal/flow/manager/manager.go
+++ b/runtime/internal/flow/manager/manager.go
@@ -310,9 +310,6 @@
// If we are dialing out to a Proxy, we need to dial a conn on this flow, and
// return a flow on that corresponding conn.
if remote.RoutingID() != c.RemoteEndpoint().RoutingID() {
- if err := sendRouteInfo(remote, f); err != nil {
- return nil, flow.NewErrDialFailed(ctx, err)
- }
c, err = conn.NewDialed(
ctx,
closer{f},
@@ -332,12 +329,6 @@
return f, nil
}
-func sendRouteInfo(ep naming.Endpoint, f flow.Flow) error {
- // TODO(suharshs): Also send endpoint routes here when implementing multi proxy.
- rid := ep.RoutingID()
- return vom.NewEncoder(f).Encode(rid.String())
-}
-
// RoutingID returns the naming.Routing of the flow.Manager.
func (m *manager) RoutingID() naming.RoutingID {
return m.rid
diff --git a/services/xproxyd/errors.vdl b/services/xproxyd/errors.vdl
index f0a32c4..0a2dbd5 100644
--- a/services/xproxyd/errors.vdl
+++ b/services/xproxyd/errors.vdl
@@ -6,4 +6,5 @@
error (
NotListening() {"en": "Proxy is not listening on any endpoints."}
+ UnexpectedMessage(msgType string) {"en": "Unexpected message of type{:msgType}"}
)
\ No newline at end of file
diff --git a/services/xproxyd/errors.vdl.go b/services/xproxyd/errors.vdl.go
index c1be22c..688e8ad 100644
--- a/services/xproxyd/errors.vdl.go
+++ b/services/xproxyd/errors.vdl.go
@@ -15,14 +15,21 @@
)
var (
- ErrNotListening = verror.Register("v.io/x/ref/services/xproxyd.NotListening", verror.NoRetry, "{1:}{2:} Proxy is not listening on any endpoints.")
+ ErrNotListening = verror.Register("v.io/x/ref/services/xproxyd.NotListening", verror.NoRetry, "{1:}{2:} Proxy is not listening on any endpoints.")
+ ErrUnexpectedMessage = verror.Register("v.io/x/ref/services/xproxyd.UnexpectedMessage", verror.NoRetry, "{1:}{2:} Unexpected message of type{:3}")
)
func init() {
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrNotListening.ID), "{1:}{2:} Proxy is not listening on any endpoints.")
+ i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrUnexpectedMessage.ID), "{1:}{2:} Unexpected message of type{:3}")
}
// NewErrNotListening returns an error with the ErrNotListening ID.
func NewErrNotListening(ctx *context.T) error {
return verror.New(ErrNotListening, ctx)
}
+
+// NewErrUnexpectedMessage returns an error with the ErrUnexpectedMessage ID.
+func NewErrUnexpectedMessage(ctx *context.T, msgType string) error {
+ return verror.New(ErrUnexpectedMessage, ctx, msgType)
+}
diff --git a/services/xproxyd/proxyd.go b/services/xproxyd/proxyd.go
index 6bd5183..6857ec8 100644
--- a/services/xproxyd/proxyd.go
+++ b/services/xproxyd/proxyd.go
@@ -5,11 +5,13 @@
package xproxyd
import (
+ "fmt"
"io"
"v.io/v23"
"v.io/v23/context"
"v.io/v23/flow"
+ "v.io/v23/flow/message"
"v.io/v23/naming"
"v.io/v23/security"
"v.io/v23/vom"
@@ -26,7 +28,6 @@
m: v23.ExperimentalGetFlowManager(ctx),
}
for _, addr := range v23.GetListenSpec(ctx).Addrs {
- ctx.Infof("proxy listening on %v", addr)
if err := p.m.Listen(ctx, addr.Protocol, addr.Address); err != nil {
return nil, err
}
@@ -54,16 +55,7 @@
}
func (p *proxy) startRouting(ctx *context.T, f flow.Flow) error {
- rid, err := p.readRouteInfo(ctx, f)
- if err != nil {
- return err
- }
- // TODO(suharshs): Find a better way to do this.
- ep, err := v23.NewEndpoint("@6@@@@" + rid.String() + "@@@@")
- if err != nil {
- return err
- }
- fout, err := p.m.Dial(ctx, ep, proxyBlessingsForPeer{}.run)
+ fout, err := p.dialNextHop(ctx, f)
if err != nil {
return err
}
@@ -72,14 +64,6 @@
return nil
}
-type proxyBlessingsForPeer struct{}
-
-// TODO(suharshs): Figure out what blessings to present here. And present discharges.
-func (proxyBlessingsForPeer) run(ctx *context.T, lep, rep naming.Endpoint, rb security.Blessings,
- rd map[string]security.Discharge) (security.Blessings, map[string]security.Discharge, error) {
- return v23.GetPrincipal(ctx).BlessingStore().Default(), nil, nil
-}
-
func (p *proxy) replyToServer(ctx *context.T, f flow.Flow) error {
eps := p.ListeningEndpoints()
if len(eps) == 0 {
@@ -113,20 +97,55 @@
}
}
-func (p *proxy) readRouteInfo(ctx *context.T, f flow.Flow) (naming.RoutingID, error) {
+func (p *proxy) dialNextHop(ctx *context.T, f flow.Flow) (flow.Flow, error) {
// TODO(suharshs): Read route information here when implementing multi proxy.
- var (
- rid string
- ret naming.RoutingID
- )
- if err := vom.NewDecoder(f).Decode(&rid); err != nil {
- return ret, err
+ m, err := readSetupMessage(ctx, f)
+ if err != nil {
+ return nil, err
}
- err := ret.FromString(rid)
- return ret, err
+ fout, err := p.m.Dial(ctx, m.PeerRemoteEndpoint, proxyBlessingsForPeer{}.run)
+ if err != nil {
+ return nil, err
+ }
+ // Write the setup message back onto the flow for the next hop to read.
+ return fout, writeSetupMessage(ctx, m, fout)
+}
+
+func readSetupMessage(ctx *context.T, f flow.Flow) (*message.Setup, error) {
+ b, err := f.ReadMsg()
+ if err != nil {
+ return nil, err
+ }
+ m, err := message.Read(ctx, b)
+ if err != nil {
+ return nil, err
+ }
+ if m, isSetup := m.(*message.Setup); isSetup {
+ return m, nil
+ }
+ return nil, NewErrUnexpectedMessage(ctx, fmt.Sprintf("%t", m))
+}
+
+func writeSetupMessage(ctx *context.T, m message.Message, f flow.Flow) error {
+ // TODO(suharshs): When reading the routes we should remove the read route from
+ // the endpoint.
+ w, err := message.Append(ctx, m, []byte{})
+ if err != nil {
+ return err
+ }
+ _, err = f.WriteMsg(w)
+ return err
}
func (p *proxy) shouldRoute(f flow.Flow) bool {
rid := f.Conn().LocalEndpoint().RoutingID()
return rid != p.m.RoutingID() && rid != naming.NullRoutingID
}
+
+type proxyBlessingsForPeer struct{}
+
+// TODO(suharshs): Figure out what blessings to present here. And present discharges.
+func (proxyBlessingsForPeer) run(ctx *context.T, lep, rep naming.Endpoint, rb security.Blessings,
+ rd map[string]security.Discharge) (security.Blessings, map[string]security.Discharge, error) {
+ return v23.GetPrincipal(ctx).BlessingStore().Default(), nil, nil
+}
