veyron/runtimes/google/ipc: add options to ipc server constructor to control
which endpoints to mount when publishing, and to allow for endpoint rewrite.
Details:
- the new ServerPublishOpt option is a server opt and offers the choices of
mounting all listened endpoints (the default), or mounting the first endpoint
only.
- the new EndpointRewriteOpt is a listener opt that tells the listener whether to
massage the endpoint it returns to replace a tcp ip with a desired host or
ip (this is needed for e.g. GCE where the public ip that we want to put in
the mounttable differs from the ip that the process sees when listening).
Change-Id: I0c870399a27a9201277882abcb6a2e860a065f4c
diff --git a/runtimes/google/ipc/full_test.go b/runtimes/google/ipc/full_test.go
index 6f2fc69..252d3e3 100644
--- a/runtimes/google/ipc/full_test.go
+++ b/runtimes/google/ipc/full_test.go
@@ -5,6 +5,7 @@
"fmt"
"io"
"log"
+ "net"
"reflect"
"strings"
"sync"
@@ -15,6 +16,7 @@
imanager "veyron/runtimes/google/ipc/stream/manager"
"veyron/runtimes/google/ipc/stream/vc"
"veyron/runtimes/google/ipc/version"
+ inaming "veyron/runtimes/google/naming"
isecurity "veyron/runtimes/google/security"
icaveat "veyron/runtimes/google/security/caveat"
@@ -337,6 +339,7 @@
client, err := InternalNewClient(mgr, mt, veyron2.LocalID(test.clientID))
if err != nil {
t.Errorf("%s: Client creation failed: %v", name, err)
+ stopServer(t, server, mt)
continue
}
if _, err := client.StartCall("mountpoint/server/suffix", "irrelevant", nil, veyron2.RemoteID(test.pattern)); !matchesErrorPattern(err, test.err) {
@@ -676,6 +679,70 @@
}
}
+// TestPublishOptions verifies that the options that are relevant to how
+// a server publishes its endpoints have the right effect.
+func TestPublishOptions(t *testing.T) {
+ sm := imanager.InternalNew(naming.FixedRoutingID(0x555555555))
+ mt := newMountTable()
+ cases := []struct {
+ opts []ipc.ServerOpt
+ expect []string
+ }{
+ {[]ipc.ServerOpt{}, []string{"127.0.0.1", "127.0.0.1"}},
+ {[]ipc.ServerOpt{veyron2.PublishAll}, []string{"127.0.0.1", "127.0.0.1"}},
+ {[]ipc.ServerOpt{veyron2.PublishFirst}, []string{"127.0.0.1"}},
+ {[]ipc.ServerOpt{veyron2.EndpointRewriteOpt("example.com")}, []string{"example.com", "example.com"}},
+ {[]ipc.ServerOpt{veyron2.PublishFirst, veyron2.EndpointRewriteOpt("example.com")}, []string{"example.com"}},
+ }
+ for i, c := range cases {
+ server, err := InternalNewServer(sm, mt, append([]ipc.ServerOpt{veyron2.LocalID(serverID)}, c.opts...)...)
+ if err != nil {
+ t.Errorf("InternalNewServer failed: %v", err)
+ continue
+ }
+ if _, err := server.Listen("tcp", "localhost:0"); err != nil {
+ t.Errorf("server.Listen failed: %v", err)
+ server.Stop()
+ continue
+ }
+ if _, err := server.Listen("tcp", "localhost:0"); err != nil {
+ t.Errorf("server.Listen failed: %v", err)
+ server.Stop()
+ continue
+ }
+ if err := server.Publish("mountpoint"); err != nil {
+ t.Errorf("server.Publish failed: %v", err)
+ server.Stop()
+ continue
+ }
+ servers, err := mt.Resolve("mountpoint")
+ if err != nil {
+ t.Errorf("mountpoint not found in mounttable")
+ server.Stop()
+ continue
+ }
+ var got []string
+ for _, s := range servers {
+ address, _ := naming.SplitAddressName(s)
+ ep, err := inaming.NewEndpoint(address)
+ if err != nil {
+ t.Errorf("case #%d: server with invalid endpoint %q: %v", i, address, err)
+ continue
+ }
+ host, _, err := net.SplitHostPort(ep.Addr().String())
+ if err != nil {
+ t.Errorf("case #%d: server endpoint with invalid address %q: %v", i, ep.Addr(), err)
+ continue
+ }
+ got = append(got, host)
+ }
+ if want := c.expect; !reflect.DeepEqual(want, got) {
+ t.Errorf("case #%d: expected mounted servers with addresses %q, got %q instead", i, want, got)
+ }
+ server.Stop()
+ }
+}
+
func init() {
var err error
if clientID, err = isecurity.NewPrivateID("client"); err != nil {
diff --git a/runtimes/google/ipc/server.go b/runtimes/google/ipc/server.go
index ca493b4..bcac1e3 100644
--- a/runtimes/google/ipc/server.go
+++ b/runtimes/google/ipc/server.go
@@ -11,6 +11,7 @@
isecurity "veyron/runtimes/google/security"
"veyron/runtimes/google/security/wire"
+ "veyron2"
"veyron2/ipc"
"veyron2/ipc/stream"
"veyron2/naming"
@@ -40,6 +41,7 @@
active sync.WaitGroup // active goroutines we've spawned.
stopped bool // whether the server has been stopped.
mt naming.MountTable
+ publishOpt veyron2.ServerPublishOpt // which endpoints to publish
}
func InternalNewServer(streamMgr stream.Manager, mt naming.MountTable, opts ...ipc.ServerOpt) (ipc.Server, error) {
@@ -49,10 +51,13 @@
publisher: InternalNewPublisher(mt, publishPeriod),
mt: mt,
}
- // Collect all ServerOpts that are also ListenerOpts.
for _, opt := range opts {
- if lopt, ok := opt.(stream.ListenerOpt); ok {
- s.listenerOpts = append(s.listenerOpts, lopt)
+ switch opt := opt.(type) {
+ case stream.ListenerOpt:
+ // Collect all ServerOpts that are also ListenerOpts.
+ s.listenerOpts = append(s.listenerOpts, opt)
+ case veyron2.ServerPublishOpt:
+ s.publishOpt = opt
}
}
return s, nil
@@ -147,8 +152,14 @@
}(flow)
}
}(ln, ep)
+ var publishEP string
+ if s.publishOpt == veyron2.PublishAll || len(s.listeners) == 1 {
+ publishEP = naming.JoinAddressName(ep.String(), "")
+ }
s.Unlock()
- s.publisher.AddServer("/" + ep.String())
+ if len(publishEP) > 0 {
+ s.publisher.AddServer(publishEP)
+ }
return ep, nil
}
diff --git a/runtimes/google/ipc/stream/manager/manager.go b/runtimes/google/ipc/stream/manager/manager.go
index 6638aaa..3b1a084 100644
--- a/runtimes/google/ipc/stream/manager/manager.go
+++ b/runtimes/google/ipc/stream/manager/manager.go
@@ -14,6 +14,7 @@
"veyron/runtimes/google/ipc/version"
inaming "veyron/runtimes/google/naming"
+ "veyron2"
"veyron2/ipc/stream"
"veyron2/naming"
"veyron2/verror"
@@ -108,6 +109,14 @@
}
func (m *manager) Listen(protocol, address string, opts ...stream.ListenerOpt) (stream.Listener, naming.Endpoint, error) {
+ var rewriteEP string
+ for i, o := range opts {
+ if rewriteOpt, ok := o.(veyron2.EndpointRewriteOpt); ok {
+ // Last one 'wins'.
+ rewriteEP = string(rewriteOpt)
+ opts = append(opts[:i], opts[i+1:]...)
+ }
+ }
m.muListeners.Lock()
if m.shutdown {
m.muListeners.Unlock()
@@ -138,7 +147,15 @@
m.listeners[ln] = true
m.muListeners.Unlock()
- ep := version.Endpoint(netln.Addr().Network(), netln.Addr().String(), m.rid)
+ network, address := netln.Addr().Network(), netln.Addr().String()
+ if network == "tcp" && len(rewriteEP) > 0 {
+ if _, port, err := net.SplitHostPort(address); err != nil {
+ return nil, nil, fmt.Errorf("%q not a valid address: %v", address, err)
+ } else {
+ address = net.JoinHostPort(rewriteEP, port)
+ }
+ }
+ ep := version.Endpoint(network, address, m.rid)
return ln, ep, nil
}
