Merge pull request #95 from aanand/server-dsl

backends/aggregate.go

@@ -11,22 +11,21 @@ import (
 
 func Aggregate() beam.Sender {
 	backend := beam.NewServer()
-	backend.OnSpawn(beam.Handler(func(ctx *beam.Message) error {
+	backend.OnSpawn(func(cmd ...string) (beam.Sender, error) {
 		allBackends := New()
 		instance := beam.NewServer()
 
-		a, err := newAggregator(allBackends, instance, ctx.Args)
+		a, err := newAggregator(allBackends, instance, cmd)
 		if err != nil {
-			return err
+			return nil, err
 		}
 
-		instance.OnAttach(beam.Handler(a.attach))
+		instance.OnAttach(a.attach)
-		instance.OnStart(beam.Handler(a.start))
+		instance.OnStart(a.start)
-		instance.OnLs(beam.Handler(a.ls))
+		instance.OnLs(a.ls)
 
-		_, err = ctx.Ret.Send(&beam.Message{Verb: beam.Ack, Ret: instance})
+		return instance, nil
-		return err
+	})
-	}))
 	return backend
 }
 
@@ -61,13 +60,13 @@ func newAggregator(allBackends *beam.Object, server *beam.Server, args []string)
 	return a, nil
 }
 
-func (a *aggregator) attach(ctx *beam.Message) error {
+func (a *aggregator) attach(name string, ret beam.Sender) error {
-	if ctx.Args[0] != "" {
+	if name != "" {
 		// TODO: implement this?
 		return fmt.Errorf("attaching to a child is not implemented")
 	}
 
-	if _, err := ctx.Ret.Send(&beam.Message{Verb: beam.Ack, Ret: a.server}); err != nil {
+	if _, err := ret.Send(&beam.Message{Verb: beam.Ack, Ret: a.server}); err != nil {
 		return err
 	}
 
@@ -81,7 +80,7 @@ func (a *aggregator) attach(ctx *beam.Message) error {
 		copies.Add(1)
 		go func() {
 			log.Printf("copying output from %#v\n", b)
-			beam.Copy(ctx.Ret, r)
+			beam.Copy(ret, r)
 			log.Printf("finished output from %#v\n", b)
 			copies.Done()
 		}()
@@ -91,29 +90,26 @@ func (a *aggregator) attach(ctx *beam.Message) error {
 	return nil
 }
 
-func (a *aggregator) start(ctx *beam.Message) error {
+func (a *aggregator) start() error {
 	for _, b := range a.backends {
 		err := b.Start()
 		if err != nil {
 			return err
 		}
 	}
-	_, err := ctx.Ret.Send(&beam.Message{Verb: beam.Ack})
+	return nil
-	return err
 }
 
-func (a *aggregator) ls(ctx *beam.Message) error {
+func (a *aggregator) ls() ([]string, error) {
 	var children []string
 
 	for _, b := range a.backends {
 		bChildren, err := b.Ls()
 		if err != nil {
-			return err
+			return nil, err
 		}
 		children = append(children, bChildren...)
 	}
 
-	ctx.Ret.Send(&beam.Message{Verb: beam.Set, Args: children})
+	return children, nil
-
-	return nil
 }

backends/dockerclient.go

@@ -31,26 +31,25 @@ func DockerClient() beam.Sender {
 
 func DockerClientWithConfig(config *DockerClientConfig) beam.Sender {
 	backend := beam.NewServer()
-	backend.OnSpawn(beam.Handler(func(ctx *beam.Message) error {
+	backend.OnSpawn(func(cmd ...string) (beam.Sender, error) {
-		if len(ctx.Args) != 1 {
+		if len(cmd) != 1 {
-			return fmt.Errorf("dockerclient: spawn takes exactly 1 argument, got %d", len(ctx.Args))
+			return nil, fmt.Errorf("dockerclient: spawn takes exactly 1 argument, got %d", len(cmd))
 		}
 		client := newClient()
 		client.scheme = config.Scheme
 		client.urlHost = config.URLHost
 		client.transport.TLSClientConfig = config.TLSClientConfig
-		client.setURL(ctx.Args[0])
+		client.setURL(cmd[0])
 		b := &dockerClientBackend{
 			client: client,
 			Server: beam.NewServer(),
 		}
-		b.Server.OnAttach(beam.Handler(b.attach))
+		b.Server.OnAttach(b.attach)
-		b.Server.OnStart(beam.Handler(b.start))
+		b.Server.OnStart(b.start)
-		b.Server.OnLs(beam.Handler(b.ls))
+		b.Server.OnLs(b.ls)
-		b.Server.OnSpawn(beam.Handler(b.spawn))
+		b.Server.OnSpawn(b.spawn)
-		_, err := ctx.Ret.Send(&beam.Message{Verb: beam.Ack, Ret: b.Server})
+		return b.Server, nil
-		return err
+	})
-	}))
 	return backend
 }
 
@@ -59,12 +58,12 @@ type dockerClientBackend struct {
 	*beam.Server
 }
 
-func (b *dockerClientBackend) attach(ctx *beam.Message) error {
+func (b *dockerClientBackend) attach(name string, ret beam.Sender) error {
-	if ctx.Args[0] == "" {
+	if name == "" {
-		ctx.Ret.Send(&beam.Message{Verb: beam.Ack, Ret: b.Server})
+		ret.Send(&beam.Message{Verb: beam.Ack, Ret: b.Server})
 		<-make(chan struct{})
 	} else {
-		path := fmt.Sprintf("/containers/%s/json", ctx.Args[0])
+		path := fmt.Sprintf("/containers/%s/json", name)
 		resp, err := b.client.call("GET", path, "")
 		if err != nil {
 			return err
@@ -76,74 +75,66 @@ func (b *dockerClientBackend) attach(ctx *beam.Message) error {
 		if resp.StatusCode != 200 {
 			return fmt.Errorf("%s", respBody)
 		}
-		c := b.newContainer(ctx.Args[0])
+		c := b.newContainer(name)
-		ctx.Ret.Send(&beam.Message{Verb: beam.Ack, Ret: c})
+		ret.Send(&beam.Message{Verb: beam.Ack, Ret: c})
 	}
 	return nil
 }
 
-func (b *dockerClientBackend) start(ctx *beam.Message) error {
+func (b *dockerClientBackend) start() error {
-	ctx.Ret.Send(&beam.Message{Verb: beam.Ack})
 	return nil
 }
 
-func (b *dockerClientBackend) ls(ctx *beam.Message) error {
+func (b *dockerClientBackend) ls() ([]string, error) {
 	resp, err := b.client.call("GET", "/containers/json", "")
 	if err != nil {
-		return fmt.Errorf("get: %v", err)
+		return nil, fmt.Errorf("get: %v", err)
 	}
 	// FIXME: check for response error
 	c := engine.NewTable("Created", 0)
 	body, err := ioutil.ReadAll(resp.Body)
 	if err != nil {
-		return fmt.Errorf("read body: %v", err)
+		return nil, fmt.Errorf("read body: %v", err)
 	}
 	if _, err := c.ReadListFrom(body); err != nil {
-		return fmt.Errorf("readlist: %v", err)
+		return nil, fmt.Errorf("readlist: %v", err)
 	}
 	names := []string{}
 	for _, env := range c.Data {
 		names = append(names, env.GetList("Names")[0][1:])
 	}
-	if _, err := ctx.Ret.Send(&beam.Message{Verb: beam.Set, Args: names}); err != nil {
+	return names, nil
-		return fmt.Errorf("send response: %v", err)
-	}
-	return nil
 }
 
-func (b *dockerClientBackend) spawn(ctx *beam.Message) error {
+func (b *dockerClientBackend) spawn(cmd ...string) (beam.Sender, error) {
-	if len(ctx.Args) != 1 {
+	if len(cmd) != 1 {
-		return fmt.Errorf("dockerclient: spawn takes exactly 1 argument, got %d", len(ctx.Args))
+		return nil, fmt.Errorf("dockerclient: spawn takes exactly 1 argument, got %d", len(cmd))
 	}
-	resp, err := b.client.call("POST", "/containers/create", ctx.Args[0])
+	resp, err := b.client.call("POST", "/containers/create", cmd[0])
 	if err != nil {
-		return err
+		return nil, err
 	}
 	respBody, err := ioutil.ReadAll(resp.Body)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	if resp.StatusCode != 201 {
-		return fmt.Errorf("expected status code 201, got %d:\n%s", resp.StatusCode, respBody)
+		return nil, fmt.Errorf("expected status code 201, got %d:\n%s", resp.StatusCode, respBody)
 	}
 	var respJson struct{ Id string }
 	if err = json.Unmarshal(respBody, &respJson); err != nil {
-		return err
+		return nil, err
 	}
-	c := b.newContainer(respJson.Id)
+	return b.newContainer(respJson.Id), nil
-	if _, err = ctx.Ret.Send(&beam.Message{Verb: beam.Ack, Ret: c}); err != nil {
-		return err
-	}
-	return nil
 }
 
 func (b *dockerClientBackend) newContainer(id string) beam.Sender {
 	c := &container{backend: b, id: id}
 	instance := beam.NewServer()
-	instance.OnAttach(beam.Handler(c.attach))
+	instance.OnAttach(c.attach)
-	instance.OnStart(beam.Handler(c.start))
+	instance.OnStart(c.start)
-	instance.OnStop(beam.Handler(c.stop))
+	instance.OnStop(c.stop)
-	instance.OnGet(beam.Handler(c.get))
+	instance.OnGet(c.get)
 	return instance
 }
 
@@ -152,8 +143,8 @@ type container struct {
 	id      string
 }
 
-func (c *container) attach(ctx *beam.Message) error {
+func (c *container) attach(name string, ret beam.Sender) error {
-	if _, err := ctx.Ret.Send(&beam.Message{Verb: beam.Ack}); err != nil {
+	if _, err := ret.Send(&beam.Message{Verb: beam.Ack}); err != nil {
 		return err
 	}
 
@@ -161,14 +152,14 @@ func (c *container) attach(ctx *beam.Message) error {
 
 	stdoutR, stdoutW := io.Pipe()
 	stderrR, stderrW := io.Pipe()
-	go beam.EncodeStream(ctx.Ret, stdoutR, "stdout")
+	go beam.EncodeStream(ret, stdoutR, "stdout")
-	go beam.EncodeStream(ctx.Ret, stderrR, "stderr")
+	go beam.EncodeStream(ret, stderrR, "stderr")
 	c.backend.client.hijack("POST", path, nil, stdoutW, stderrW)
 
 	return nil
 }
 
-func (c *container) start(ctx *beam.Message) error {
+func (c *container) start() error {
 	path := fmt.Sprintf("/containers/%s/start", c.id)
 	resp, err := c.backend.client.call("POST", path, "{}")
 	if err != nil {
@@ -181,13 +172,10 @@ func (c *container) start(ctx *beam.Message) error {
 	if resp.StatusCode != 204 {
 		return fmt.Errorf("expected status code 204, got %d:\n%s", resp.StatusCode, respBody)
 	}
-	if _, err := ctx.Ret.Send(&beam.Message{Verb: beam.Ack}); err != nil {
-		return err
-	}
 	return nil
 }
 
-func (c *container) stop(ctx *beam.Message) error {
+func (c *container) stop() error {
 	path := fmt.Sprintf("/containers/%s/stop", c.id)
 	resp, err := c.backend.client.call("POST", path, "")
 	if err != nil {
@@ -200,29 +188,23 @@ func (c *container) stop(ctx *beam.Message) error {
 	if resp.StatusCode != 204 {
 		return fmt.Errorf("expected status code 204, got %d:\n%s", resp.StatusCode, respBody)
 	}
-	if _, err := ctx.Ret.Send(&beam.Message{Verb: beam.Ack}); err != nil {
-		return err
-	}
 	return nil
 }
 
-func (c *container) get(ctx *beam.Message) error {
+func (c *container) get() (string, error) {
 	path := fmt.Sprintf("/containers/%s/json", c.id)
 	resp, err := c.backend.client.call("GET", path, "")
 	if err != nil {
-		return err
+		return "", err
 	}
 	respBody, err := ioutil.ReadAll(resp.Body)
 	if err != nil {
-		return err
+		return "", err
 	}
 	if resp.StatusCode != 200 {
-		return fmt.Errorf("%s", respBody)
+		return "", fmt.Errorf("%s", respBody)
 	}
-	if _, err := ctx.Ret.Send(&beam.Message{Verb: beam.Set, Args: []string{string(respBody)}}); err != nil {
+	return string(respBody), nil
-		return err
-	}
-	return nil
 }
 
 type client struct {

backends/dockerserver.go

@@ -22,7 +22,7 @@ import (
 
 func DockerServer() beam.Sender {
 	backend := beam.NewServer()
-	backend.OnSpawn(beam.Handler(func(ctx *beam.Message) error {
+	backend.OnVerb(beam.Spawn, beam.Handler(func(ctx *beam.Message) error {
 		instance := beam.Task(func(in beam.Receiver, out beam.Sender) {
 			url := "tcp://localhost:4243"
 			if len(ctx.Args) > 0 {

backends/exec.go

@@ -14,7 +14,7 @@ import (
 
 func Exec() beam.Sender {
 	e := beam.NewServer()
-	e.OnSpawn(beam.Handler(func(msg *beam.Message) error {
+	e.OnVerb(beam.Spawn, beam.Handler(func(msg *beam.Message) error {
 		if len(msg.Args) < 1 {
 			return fmt.Errorf("usage: SPAWN exec|... <config>")
 		}
@@ -33,7 +33,7 @@ func Exec() beam.Sender {
 			Cmd:	exec.Command(config.Path, config.Args...),
 			Server: beam.NewServer(),
 		}
-		cmd.OnAttach(beam.Handler(func(msg *beam.Message) error {
+		cmd.OnVerb(beam.Attach, beam.Handler(func(msg *beam.Message) error {
 			stdout, err := cmd.StdoutPipe()
 			if err != nil {
 				return err
@@ -76,7 +76,7 @@ func Exec() beam.Sender {
 			cmd.tasks.Wait()
 			return nil
 		}))
-		cmd.OnStart(beam.Handler(func(msg *beam.Message) error {
+		cmd.OnVerb(beam.Start, beam.Handler(func(msg *beam.Message) error {
 			cmd.tasks.Add(1)
 			if err := cmd.Cmd.Start(); err != nil {
 				return err

backends/fakeclient.go

@@ -9,7 +9,7 @@ import (
 
 func FakeClient() beam.Sender {
 	backend := beam.NewServer()
-	backend.OnSpawn(beam.Handler(func(ctx *beam.Message) error {
+	backend.OnVerb(beam.Spawn, beam.Handler(func(ctx *beam.Message) error {
 		// Instantiate a new fakeclient instance
 		instance := beam.Task(func(in beam.Receiver, out beam.Sender) {
 			fmt.Printf("fake client!\n")

backends/orchard.go

@@ -13,11 +13,11 @@ import (
 
 func Orchard() beam.Sender {
 	backend := beam.NewServer()
-	backend.OnSpawn(beam.Handler(func(ctx *beam.Message) error {
+	backend.OnSpawn(func(cmd ...string) (beam.Sender, error) {
-		if len(ctx.Args) != 2 {
+		if len(cmd) != 2 {
-			return fmt.Errorf("orchard: spawn expects 2 arguments: API token and name of host")
+			return nil, fmt.Errorf("orchard: spawn expects 2 arguments: API token and name of host")
 		}
-		apiToken, hostName := ctx.Args[0], ctx.Args[1]
+		apiToken, hostName := cmd[0], cmd[1]
 
 		apiClient := &api.HTTPClient{
 			BaseURL: "https://api.orchardup.com/v2",
@@ -26,13 +26,13 @@ func Orchard() beam.Sender {
 
 		host, err := apiClient.GetHost(hostName)
 		if err != nil {
-			return err
+			return nil, err
 		}
 
 		url := fmt.Sprintf("tcp://%s:4243", host.IPAddress)
 		tlsConfig, err := getTLSConfig([]byte(host.ClientCert), []byte(host.ClientKey))
 		if err != nil {
-			return err
+			return nil, err
 		}
 
 		backend := DockerClientWithConfig(&DockerClientConfig{
@@ -43,12 +43,11 @@ func Orchard() beam.Sender {
 		forwardBackend := beam.Obj(backend)
 		forwardInstance, err := forwardBackend.Spawn(url)
 		if err != nil {
-			return err
+			return nil, err
 		}
 
-		_, err = ctx.Ret.Send(&beam.Message{Verb: beam.Ack, Ret: forwardInstance})
+		return forwardInstance, nil
-		return err
+	})
-	}))
 	return backend
 }
 

backends/shipyard.go

@@ -13,7 +13,7 @@ import (
 
 func Shipyard() beam.Sender {
 	backend := beam.NewServer()
-	backend.OnSpawn(beam.Handler(func(ctx *beam.Message) error {
+	backend.OnVerb(beam.Spawn, beam.Handler(func(ctx *beam.Message) error {
 		if len(ctx.Args) != 3 {
 			return fmt.Errorf("Shipyard: Usage <shipyard URL> <user> <pass>")
 		}
@@ -21,10 +21,10 @@ func Shipyard() beam.Sender {
 		c := &shipyard{url: ctx.Args[0], user: ctx.Args[1], pass: ctx.Args[2]}
 
 		c.Server = beam.NewServer()
-		c.Server.OnAttach(beam.Handler(c.attach))
+		c.Server.OnVerb(beam.Attach, beam.Handler(c.attach))
-		c.Server.OnStart(beam.Handler(c.start))
+		c.Server.OnVerb(beam.Start, beam.Handler(c.start))
-		c.Server.OnLs(beam.Handler(c.containers))
+		c.Server.OnVerb(beam.Ls, beam.Handler(c.containers))
-		c.OnGet(beam.Handler(c.containerInspect))
+		c.OnVerb(beam.Get, beam.Handler(c.containerInspect))
 		_, err := ctx.Ret.Send(&beam.Message{Verb: beam.Ack, Ret: c.Server})
 		return err
 	}))

backends/simulator.go

@@ -6,12 +6,12 @@ import (
 
 func Simulator() beam.Sender {
 	s := beam.NewServer()
-	s.OnSpawn(beam.Handler(func(ctx *beam.Message) error {
+	s.OnVerb(beam.Spawn, beam.Handler(func(ctx *beam.Message) error {
 		containers := ctx.Args
 		instance := beam.Task(func(in beam.Receiver, out beam.Sender) {
 			beam.Obj(out).Log("[simulator] starting\n")
 			s := beam.NewServer()
-			s.OnLs(beam.Handler(func(msg *beam.Message) error {
+			s.OnVerb(beam.Ls, beam.Handler(func(msg *beam.Message) error {
 				beam.Obj(out).Log("[simulator] generating fake list of objects...\n")
 				beam.Obj(msg.Ret).Set(containers...)
 				return nil

backends/tutum.go

@@ -20,7 +20,7 @@ var (
 
 func Tutum() beam.Sender {
 	backend := beam.NewServer()
-	backend.OnSpawn(beam.Handler(func(ctx *beam.Message) error {
+	backend.OnVerb(beam.Spawn, beam.Handler(func(ctx *beam.Message) error {
 		if len(ctx.Args) == 2 {
 			tutum.User = ctx.Args[0]
 			tutum.ApiKey = ctx.Args[1]
@@ -36,10 +36,10 @@ func Tutum() beam.Sender {
 			tutumDockerConnector: tutumDockerConnector,
 			Server:               beam.NewServer(),
 		}
-		t.Server.OnAttach(beam.Handler(t.attach))
+		t.Server.OnVerb(beam.Attach, beam.Handler(t.attach))
-		t.Server.OnStart(beam.Handler(t.ack))
+		t.Server.OnVerb(beam.Start, beam.Handler(t.ack))
-		t.Server.OnLs(beam.Handler(t.ls))
+		t.Server.OnVerb(beam.Ls, beam.Handler(t.ls))
-		t.Server.OnSpawn(beam.Handler(t.spawn))
+		t.Server.OnVerb(beam.Spawn, beam.Handler(t.spawn))
 		_, err = ctx.Ret.Send(&beam.Message{Verb: beam.Ack, Ret: t.Server})
 		return err
 	}))
@@ -121,9 +121,9 @@ func (t *tutumBackend) spawn(ctx *beam.Message) error {
 func (t *tutumBackend) newContainer(id string) beam.Sender {
 	c := &tutumContainer{tutumBackend: t, id: id}
 	instance := beam.NewServer()
-	instance.OnGet(beam.Handler(c.get))
+	instance.OnVerb(beam.Get, beam.Handler(c.get))
-	instance.OnStart(beam.Handler(c.start))
+	instance.OnVerb(beam.Start, beam.Handler(c.start))
-	instance.OnStop(beam.Handler(c.stop))
+	instance.OnVerb(beam.Stop, beam.Handler(c.stop))
 	return instance
 }
 

beam/server.go

@@ -27,36 +27,75 @@ func (s *Server) OnVerb(v Verb, h Sender) *Server {
 	return s
 }
 
-func (s *Server) OnSpawn(h Sender) *Server {
+func (s *Server) OnLog(h func(...string) error) *Server {
-	return s.OnVerb(Spawn, h)
+	return s.OnVerb(Log, Handler(func(msg *Message) error {
+		return h(msg.Args...)
+	}))
 }
 
-func (s *Server) OnStart(h Sender) *Server {
+func (s *Server) OnLs(h func() ([]string, error)) *Server {
-	return s.OnVerb(Start, h)
+	return s.OnVerb(Ls, Handler(func(msg *Message) error {
+		names, err := h()
+		if err != nil {
+			return err
+		}
+		_, err = msg.Ret.Send(&Message{Verb: Set, Args: names})
+		return err
+	}))
 }
 
-func (s *Server) OnStop(h Sender) *Server {
+func (s *Server) OnSpawn(h func(cmd ...string) (Sender, error)) *Server {
-	return s.OnVerb(Stop, h)
+	return s.OnVerb(Spawn, Handler(func(msg *Message) error {
+		obj, err := h(msg.Args...)
+		if err != nil {
+			return err
+		}
+		_, err = msg.Ret.Send(&Message{Verb: Ack, Ret: obj})
+		return err
+	}))
 }
 
-func (s *Server) OnAttach(h Sender) *Server {
+func (s *Server) OnAttach(h func(name string, ret Sender) error) *Server {
-	return s.OnVerb(Attach, h)
+	return s.OnVerb(Attach, Handler(func(msg *Message) error {
+		return h(msg.Args[0], msg.Ret)
+	}))
 }
 
-func (s *Server) OnLog(h Sender) *Server {
+func (s *Server) OnError(h func(...string) error) *Server {
-	return s.OnVerb(Log, h)
+	return s.OnVerb(Error, Handler(func(msg *Message) error {
+		return h(msg.Args...)
+	}))
 }
 
-func (s *Server) OnError(h Sender) *Server {
+func (s *Server) OnGet(h func() (string, error)) *Server {
-	return s.OnVerb(Error, h)
+	return s.OnVerb(Get, Handler(func(msg *Message) error {
+		content, err := h()
+		if err != nil {
+			return err
+		}
+		_, err = msg.Ret.Send(&Message{Verb: Set, Args: []string{content}})
+		return err
+	}))
 }
 
-func (s *Server) OnLs(h Sender) *Server {
+func (s *Server) OnStart(h func() error) *Server {
-	return s.OnVerb(Ls, h)
+	return s.OnVerb(Start, Handler(func(msg *Message) error {
+		if err := h(); err != nil {
+			return err
+		}
+		_, err := msg.Ret.Send(&Message{Verb: Ack})
+		return err
+	}))
 }
 
-func (s *Server) OnGet(h Sender) *Server {
+func (s *Server) OnStop(h func() error) *Server {
-	return s.OnVerb(Get, h)
+	return s.OnVerb(Stop, Handler(func(msg *Message) error {
+		if err := h(); err != nil {
+			return err
+		}
+		_, err := msg.Ret.Send(&Message{Verb: Ack})
+		return err
+	}))
 }
 
 func (s *Server) Send(msg *Message) (Receiver, error) {

beam/task.go

@@ -11,14 +11,14 @@ func Task(f func(in Receiver, out Sender)) Sender {
 	inR, inW := Pipe()
 	outR, outW := Pipe()
 	obj := NewServer()
-	obj.OnAttach(Handler(func(msg *Message) error {
+	obj.OnVerb(Attach, Handler(func(msg *Message) error {
 		msg.Ret.Send(&Message{Verb: Ack, Ret: inW})
 		fmt.Printf("copying task output from %#v to %#v\n", outR, msg.Ret)
 		defer fmt.Printf("(DONE) copying task output from %#v to %#v\n", outR, msg.Ret)
 		Copy(msg.Ret, outR)
 		return nil
 	}))
-	obj.OnStart(Handler(func(msg *Message) error {
+	obj.OnVerb(Start, Handler(func(msg *Message) error {
 		l.RLock()
 		r := running
 		l.RUnlock()

beam/tree.go

@@ -14,7 +14,7 @@ func NewTree() *Tree {
 		Server:   NewServer(),
 		children: make(map[string]Sender),
 	}
-	t.OnAttach(Handler(func(msg *Message) error {
+	t.OnVerb(Attach, Handler(func(msg *Message) error {
 		if len(msg.Args) == 0 || msg.Args[0] == "" {
 			msg.Ret.Send(&Message{Verb: Ack, Ret: t})
 			return nil
@@ -26,7 +26,7 @@ func NewTree() *Tree {
 		Obj(msg.Ret).Error("not found")
 		return nil
 	}))
-	t.OnLs(Handler(func(msg *Message) error {
+	t.OnVerb(Ls, Handler(func(msg *Message) error {
 		names := make([]string, 0, len(t.children))
 		for name := range t.children {
 			names = append(names, name)

debug/debug.go

@@ -15,7 +15,7 @@ func Debug() beam.Sender {
 	}
 
 	sender := beam.NewServer()
-	sender.OnSpawn(beam.Handler(dbgInstance.spawn))
+	sender.OnVerb(beam.Spawn, beam.Handler(dbgInstance.spawn))
 	return sender
 }
 

swarmd/swarmd.go

@@ -25,14 +25,14 @@ func main() {
 
 func cmdDaemon(c *cli.Context) {
 	app := beam.NewServer()
-	app.OnLog(beam.Handler(func(msg *beam.Message) error {
+	app.OnLog(func(args ...string) error {
-		log.Printf("%s\n", strings.Join(msg.Args, " "))
+		log.Printf("%s\n", strings.Join(args, " "))
 		return nil
-	}))
+	})
-	app.OnError(beam.Handler(func(msg *beam.Message) error {
+	app.OnError(func(args ...string) error {
-		Fatalf("Fatal: %v", strings.Join(msg.Args[:1], ""))
+		Fatalf("Fatal: %v", strings.Join(args[:1], ""))
 		return nil
-	}))
+	})
 	back := backends.New()
 	if len(c.Args()) == 0 {
 		names, err := back.Ls()