Merge pull request #32 from bfirsh/multiple-backends-pipe

backends/backends.go

@@ -24,12 +24,21 @@ func New() *beam.Object {
 func Debug() beam.Sender {
 	backend := beam.NewServer()
 	backend.OnSpawn(beam.Handler(func(ctx *beam.Message) error {
-		instance := beam.NewServer()
-		instance.Catchall(beam.Handler(func(msg *beam.Message) error {
-			fmt.Printf("[DEBUG] %s %s\n", msg.Verb, strings.Join(msg.Args, " "))
-			ctx.Ret.Send(msg)
-			return nil
-		}))
+		instance := beam.Task(func(in beam.Receiver, out beam.Sender) {
+			fmt.Printf("debug backend!")
+			for {
+				msg, err := in.Receive(beam.Ret)
+				if err != nil {
+					fmt.Printf("debug receive: %v", err)
+					return
+				}
+				fmt.Printf("[DEBUG] %s %s\n", msg.Verb, strings.Join(msg.Args, " "))
+				if _, err := out.Send(msg); err != nil {
+					fmt.Printf("debug send: %v", err)
+					return
+				}
+			}
+		})
 		_, err := ctx.Ret.Send(&beam.Message{Verb: beam.Ack, Ret: instance})
 		return err
 	}))

beam/verbs.go

@@ -17,3 +17,35 @@ const (
 	Stop
 	Watch
 )
+
+func (v Verb) String() string {
+	switch v {
+	case Ack:
+		return "Ack"
+	case Attach:
+		return "Attach"
+	case Connect:
+		return "Connect"
+	case Error:
+		return "Error"
+	case File:
+		return "File"
+	case Get:
+		return "Get"
+	case Log:
+		return "Log"
+	case Ls:
+		return "Ls"
+	case Set:
+		return "Set"
+	case Spawn:
+		return "Spawn"
+	case Start:
+		return "Start"
+	case Stop:
+		return "Stop"
+	case Watch:
+		return "Watch"
+	}
+	return ""
+}

swarmd/swarmd.go

@@ -18,8 +18,7 @@ func main() {
 	app.Name = "swarmd"
 	app.Usage = "Compose distributed systems from lightweight services"
 	app.Version = "0.0.1"
-	app.Flags = []cli.Flag{
-	}
+	app.Flags = []cli.Flag{}
 	app.Action = cmdDaemon
 	app.Run(os.Args)
 }
@@ -43,30 +42,37 @@ func cmdDaemon(c *cli.Context) {
 		fmt.Println(strings.Join(names, "\n"))
 		return
 	}
-	bName, bArgs, err := parseCmd(c.Args()[0])
-	if err != nil {
-		Fatalf("parse: %v", err)
+	var previousInstanceIn beam.Receiver
+	for _, backendArg := range c.Args() {
+		bName, bArgs, err := parseCmd(backendArg)
+		if err != nil {
+			Fatalf("parse: %v", err)
+		}
+		fmt.Printf("---> Loading backend '%s'\n", strings.Join(append([]string{bName}, bArgs...), " "))
+		_, backend, err := back.Attach(bName)
+		if err != nil {
+			Fatalf("%s: %v\n", bName, err)
+		}
+		fmt.Printf("---> Spawning\n")
+		instance, err := backend.Spawn(bArgs...)
+		if err != nil {
+			Fatalf("spawn %s: %v\n", bName, err)
+		}
+		fmt.Printf("---> Attaching\n")
+		instanceIn, instanceOut, err := instance.Attach("")
+		if err != nil {
+			Fatalf("attach: %v", err)
+		}
+		fmt.Printf("---> Starting\n")
+		if err := instance.Start(); err != nil {
+			Fatalf("start: %v", err)
+		}
+		if previousInstanceIn != nil {
+			go beam.Copy(instanceOut, previousInstanceIn)
+		}
+		previousInstanceIn = instanceIn
 	}
-	fmt.Printf("---> Loading backend '%s'\n", strings.Join(append([]string{bName}, bArgs...), " "))
-	_, backend, err := back.Attach(bName)
-	if err != nil {
-		Fatalf("%s: %v\n", bName, err)
-	}
-	fmt.Printf("---> Spawning\n")
-	instance, err := backend.Spawn(bArgs...)
-	if err != nil {
-		Fatalf("spawn %s: %v\n", bName, err)
-	}
-	fmt.Printf("---> Attaching\n")
-	instanceIn, _, err := instance.Attach("")
-	if err != nil {
-		Fatalf("attach: %v", err)
-	}
-	fmt.Printf("---> Starting\n")
-	if err := instance.Start(); err != nil {
-		Fatalf("start: %v", err)
-	}
-	_, err = beam.Copy(app, instanceIn)
+	_, err := beam.Copy(app, previousInstanceIn)
 	if err != nil {
 		Fatalf("copy: %v", err)
 	}