docker
/
docs
mirror of https://github.com/docker/docs.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Move Pipe() into the beam package 

Docker-DCO-1.1-Signed-off-by: Solomon Hykes <solomon@docker.com> (github: shykes)
This commit is contained in:
Solomon Hykes 2014-06-02 23:55:15 +00:00
parent 16b16715b2
commit 58f7d34f0f
6 changed files with 31 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
beam/inmem/inmem.go → beam/inmem.go
View File

@ -1,7 +1,6 @@
package inmem package beam
import ( import (
"github.com/docker/libswarm/beam"
"io" "io"
"sync" "sync"
) )
@ -23,10 +22,10 @@ type pipe struct {
wl sync.Mutex wl sync.Mutex
rerr error // if reader closed, error to give writes rerr error // if reader closed, error to give writes
werr error // if writer closed, error to give reads werr error // if writer closed, error to give reads
msg *beam.Message msg *Message
} }
func (p *pipe) psend(msg *beam.Message) error { func (p *pipe) psend(msg *Message) error {
var err error var err error
// One writer at a time. // One writer at a time.
p.wl.Lock() p.wl.Lock()
@ -53,16 +52,16 @@ func (p *pipe) psend(msg *beam.Message) error {
return err return err
} }
func (p *pipe) send(msg *beam.Message) (ret beam.Receiver, err error) { func (p *pipe) send(msg *Message) (ret Receiver, err error) {
// Prepare nested Receiver if requested // Prepare nested Receiver if requested
if beam.RetPipe.Equals(msg.Ret) { if RetPipe.Equals(msg.Ret) {
ret, msg.Ret = Pipe() ret, msg.Ret = Pipe()
} }
err = p.psend(msg) err = p.psend(msg)
return return
} }
func (p *pipe) preceive() (*beam.Message, error) { func (p *pipe) preceive() (*Message, error) {
p.rl.Lock() p.rl.Lock()
defer p.rl.Unlock() defer p.rl.Unlock()
@ -86,15 +85,15 @@ func (p *pipe) preceive() (*beam.Message, error) {
return msg, nil return msg, nil
} }
func (p *pipe) receive(mode int) (*beam.Message, error) { func (p *pipe) receive(mode int) (*Message, error) {
msg, err := p.preceive() msg, err := p.preceive()
if err != nil { if err != nil {
return nil, err return nil, err
} }
if msg.Ret == nil { if msg.Ret == nil {
msg.Ret = beam.NopSender{} msg.Ret = NopSender{}
} }
if mode&beam.Ret == 0 { if mode&Ret == 0 {
msg.Ret.Close() msg.Ret.Close()
} }
return msg, nil return msg, nil
@ -128,16 +127,16 @@ type PipeReceiver struct {
p *pipe p *pipe
} }
func (r *PipeReceiver) Receive(mode int) (*beam.Message, error) { func (r *PipeReceiver) Receive(mode int) (*Message, error) {
return r.p.receive(mode) return r.p.receive(mode)
} }
func (r *PipeReceiver) SendTo(dst beam.Sender) (int, error) { func (r *PipeReceiver) SendTo(dst Sender) (int, error) {
var n int var n int
// If the destination is a PipeSender, we can cheat // If the destination is a PipeSender, we can cheat
pdst, ok := dst.(*PipeSender) pdst, ok := dst.(*PipeSender)
if !ok { if !ok {
return 0, beam.ErrIncompatibleSender return 0, ErrIncompatibleSender
} }
for { for {
pmsg, err := r.p.preceive() pmsg, err := r.p.preceive()
@ -170,16 +169,16 @@ type PipeSender struct {
p *pipe p *pipe
} }
func (w *PipeSender) Send(msg *beam.Message) (beam.Receiver, error) { func (w *PipeSender) Send(msg *Message) (Receiver, error) {
return w.p.send(msg) return w.p.send(msg)
} }
func (w *PipeSender) ReceiveFrom(src beam.Receiver) (int, error) { func (w *PipeSender) ReceiveFrom(src Receiver) (int, error) {
var n int var n int
// If the destination is a PipeReceiver, we can cheat // If the destination is a PipeReceiver, we can cheat
psrc, ok := src.(*PipeReceiver) psrc, ok := src.(*PipeReceiver)
if !ok { if !ok {
return 0, beam.ErrIncompatibleReceiver return 0, ErrIncompatibleReceiver
} }
for { for {
pmsg, err := psrc.p.preceive() pmsg, err := psrc.p.preceive()

21
beam/inmem/inmem_test.go → beam/inmem_test.go
View File

@ -1,21 +1,20 @@
package inmem package beam
import ( import (
"fmt" "fmt"
"github.com/docker/libswarm/beam"
"github.com/dotcloud/docker/pkg/testutils" "github.com/dotcloud/docker/pkg/testutils"
"io/ioutil" "io/ioutil"
"os" "os"
"testing" "testing"
) )
func TestRetPipe(t *testing.T) { func TestInmemRetPipe(t *testing.T) {
r, w := Pipe() r, w := Pipe()
defer r.Close() defer r.Close()
defer w.Close() defer w.Close()
wait := make(chan struct{}) wait := make(chan struct{})
go func() { go func() {
ret, err := w.Send(&beam.Message{Name: "hello", Ret: beam.RetPipe}) ret, err := w.Send(&Message{Name: "hello", Ret: RetPipe})
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@ -28,11 +27,11 @@ func TestRetPipe(t *testing.T) {
} }
close(wait) close(wait)
}() }()
msg, err := r.Receive(beam.Ret) msg, err := r.Receive(Ret)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
if _, err := msg.Ret.Send(&beam.Message{Name: "this better not crash"}); err != nil { if _, err := msg.Ret.Send(&Message{Name: "this better not crash"}); err != nil {
t.Fatal(err) t.Fatal(err)
} }
<-wait <-wait
@ -55,7 +54,7 @@ func TestSimpleSend(t *testing.T) {
t.Fatalf("%#v", *msg) t.Fatalf("%#v", *msg)
} }
}() }()
if _, err := w.Send(&beam.Message{Name: "print", Args: []string{"hello world"}}); err != nil { if _, err := w.Send(&Message{Name: "print", Args: []string{"hello world"}}); err != nil {
t.Fatal(err) t.Fatal(err)
} }
}) })
@ -68,7 +67,7 @@ func TestSendReply(t *testing.T) {
testutils.Timeout(t, func() { testutils.Timeout(t, func() {
// Send // Send
go func() { go func() {
ret, err := w.Send(&beam.Message{Args: []string{"this is the request"}, Ret: beam.RetPipe}) ret, err := w.Send(&Message{Args: []string{"this is the request"}, Ret: RetPipe})
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@ -85,7 +84,7 @@ func TestSendReply(t *testing.T) {
} }
}() }()
// Receive a message with mode=Ret // Receive a message with mode=Ret
msg, err := r.Receive(beam.Ret) msg, err := r.Receive(Ret)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@ -96,7 +95,7 @@ func TestSendReply(t *testing.T) {
t.Fatalf("%#v", msg) t.Fatalf("%#v", msg)
} }
// Send a reply // Send a reply
_, err = msg.Ret.Send(&beam.Message{Args: []string{"this is the reply"}}) _, err = msg.Ret.Send(&Message{Args: []string{"this is the reply"}})
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@ -117,7 +116,7 @@ func TestSendFile(t *testing.T) {
tmp.Seek(0, 0) tmp.Seek(0, 0)
testutils.Timeout(t, func() { testutils.Timeout(t, func() {
go func() { go func() {
_, err := w.Send(&beam.Message{Name: "file", Args: []string{"path=" + tmp.Name()}, Att: tmp}) _, err := w.Send(&Message{Name: "file", Args: []string{"path=" + tmp.Name()}, Att: tmp})
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }

3
beam/utils/hub.go
View File

@ -3,7 +3,6 @@ package utils
import ( import (
"fmt" "fmt"
"github.com/docker/libswarm/beam" "github.com/docker/libswarm/beam"
"github.com/docker/libswarm/beam/inmem"
"io" "io"
"strings" "strings"
"sync" "sync"
@ -30,7 +29,7 @@ func (hub *Hub) Send(msg *beam.Message) (ret beam.Receiver, err error) {
fmt.Printf("[hub] received %v\n", msg) fmt.Printf("[hub] received %v\n", msg)
hIn := msg.Ret hIn := msg.Ret
if hIn == beam.RetPipe { if hIn == beam.RetPipe {
ret, hIn = inmem.Pipe() ret, hIn = beam.Pipe()
} }
// This queue guarantees that the first message received by the handler // This queue guarantees that the first message received by the handler
// is the "register" response. // is the "register" response.

5
beam/utils/queue.go
View File

@ -2,17 +2,16 @@ package utils
import ( import (
"github.com/docker/libswarm/beam" "github.com/docker/libswarm/beam"
"github.com/docker/libswarm/beam/inmem"
) )
type Queue struct { type Queue struct {
*inmem.PipeSender *beam.PipeSender
dst beam.Sender dst beam.Sender
ch chan *beam.Message ch chan *beam.Message
} }
func NewQueue(dst beam.Sender, size int) *Queue { func NewQueue(dst beam.Sender, size int) *Queue {
r, w := inmem.Pipe() r, w := beam.Pipe()
q := &Queue{ q := &Queue{
PipeSender: w, PipeSender: w,
dst: dst, dst: dst,

3
beam/utils/queue_test.go
View File

@ -2,12 +2,11 @@ package utils
import ( import (
"github.com/docker/libswarm/beam" "github.com/docker/libswarm/beam"
"github.com/docker/libswarm/beam/inmem"
"testing" "testing"
) )
func TestSendRet(t *testing.T) { func TestSendRet(t *testing.T) {
r, w := inmem.Pipe() r, w := beam.Pipe()
defer r.Close() defer r.Close()
defer w.Close() defer w.Close()
q := NewQueue(w, 1) q := NewQueue(w, 1)

5
beam/utils/stack_test.go
View File

@ -2,7 +2,6 @@ package utils
import ( import (
"github.com/docker/libswarm/beam" "github.com/docker/libswarm/beam"
"github.com/docker/libswarm/beam/inmem"
"github.com/docker/libswarm/beam/unix" "github.com/docker/libswarm/beam/unix"
"github.com/dotcloud/docker/pkg/testutils" "github.com/dotcloud/docker/pkg/testutils"
"strings" "strings"
@ -10,7 +9,7 @@ import (
) )
func TestStackWithPipe(t *testing.T) { func TestStackWithPipe(t *testing.T) {
r, w := inmem.Pipe() r, w := beam.Pipe()
defer r.Close() defer r.Close()
defer w.Close() defer w.Close()
s := NewStackSender() s := NewStackSender()
@ -107,7 +106,7 @@ func TestStackAddBad(t *testing.T) {
s := NewStackSender() s := NewStackSender()
buf := Buffer{} buf := Buffer{}
s.Add(&buf) s.Add(&buf)
r, w := inmem.Pipe() r, w := beam.Pipe()
s.Add(w) s.Add(w)
if s.Len() != 2 { if s.Len() != 2 {
t.Fatalf("%#v", s) t.Fatalf("%#v", s)