package engine import ( "fmt" "github.com/dotcloud/docker/pkg/beam" "github.com/dotcloud/docker/pkg/beam/data" "io" "os" "strconv" "sync" ) type Sender struct { beam.Sender } func NewSender(s beam.Sender) *Sender { return &Sender{s} } func (s *Sender) Install(eng *Engine) error { // FIXME: this doesn't exist yet. eng.RegisterCatchall(s.Handle) return nil } func (s *Sender) Handle(job *Job) Status { msg := data.Empty().Set("cmd", append([]string{job.Name}, job.Args...)...) peer, err := beam.SendConn(s, msg.Bytes()) if err != nil { return job.Errorf("beamsend: %v", err) } defer peer.Close() var tasks sync.WaitGroup defer tasks.Wait() r := beam.NewRouter(nil) r.NewRoute().KeyStartsWith("cmd", "log", "stdout").HasAttachment().Handler(func(p []byte, stdout *os.File) error { tasks.Add(1) io.Copy(job.Stdout, stdout) tasks.Done() return nil }) r.NewRoute().KeyStartsWith("cmd", "log", "stderr").HasAttachment().Handler(func(p []byte, stderr *os.File) error { tasks.Add(1) io.Copy(job.Stderr, stderr) tasks.Done() return nil }) r.NewRoute().KeyStartsWith("cmd", "log", "stdin").HasAttachment().Handler(func(p []byte, stdin *os.File) error { tasks.Add(1) io.Copy(stdin, job.Stdin) tasks.Done() return nil }) var status int r.NewRoute().KeyStartsWith("cmd", "status").Handler(func(p []byte, f *os.File) error { cmd := data.Message(p).Get("cmd") if len(cmd) != 2 { return fmt.Errorf("usage: %s <0-127>", cmd[0]) } s, err := strconv.ParseUint(cmd[1], 10, 8) if err != nil { return fmt.Errorf("usage: %s <0-127>", cmd[0]) } status = int(s) return nil }) if _, err := beam.Copy(r, peer); err != nil { return job.Errorf("%v", err) } return Status(status) } type Receiver struct { *Engine peer beam.Receiver } func NewReceiver(peer beam.Receiver) *Receiver { return &Receiver{Engine: New(), peer: peer} } func (rcv *Receiver) Run() error { r := beam.NewRouter(nil) r.NewRoute().KeyExists("cmd").Handler(func(p []byte, f *os.File) error { // Use the attachment as a beam return channel peer, err := beam.FileConn(f) if err != nil { f.Close() return err } f.Close() defer peer.Close() cmd := data.Message(p).Get("cmd") job := rcv.Engine.Job(cmd[0], cmd[1:]...) stdout, err := beam.SendRPipe(peer, data.Empty().Set("cmd", "log", "stdout").Bytes()) if err != nil { return err } job.Stdout.Add(stdout) stderr, err := beam.SendRPipe(peer, data.Empty().Set("cmd", "log", "stderr").Bytes()) if err != nil { return err } job.Stderr.Add(stderr) stdin, err := beam.SendWPipe(peer, data.Empty().Set("cmd", "log", "stdin").Bytes()) if err != nil { return err } job.Stdin.Add(stdin) // ignore error because we pass the raw status job.Run() err = peer.Send(data.Empty().Set("cmd", "status", fmt.Sprintf("%d", job.status)).Bytes(), nil) if err != nil { return err } return nil }) _, err := beam.Copy(r, rcv.peer) return err }