add loop socket

1 files changed, 41 insertions, 0 deletions

diff --git a/pkg/server/socket/socket.go b/pkg/server/socket/socket.go
index c91423e..4629e22 100644
--- a/pkg/server/socket/socket.go
+++ b/pkg/server/socket/socket.go
@@ -36,6 +36,12 @@ type connChannel struct {
 	once sync.Once
 }
 
+type loopSocket struct{}
+
+type loopChannel struct {
+	q queue.Q
+}
+
 func newConnChannel(conn net.Conn) Channel {
 	return &connChannel{conn: conn}
 }
@@ -139,6 +145,37 @@ func (s *dialSocket) Open(env env.Env) (Channel, error) {
 func (s *dialSocket) Close() {
 }
 
+func (c *loopChannel) Send(wq queue.Q) error {
+	return queue.Copy(c.q, wq)
+}
+
+func (c *loopChannel) Recv(rq queue.Q) error {
+	defer close(c.q)
+	return queue.Copy(rq, c.q)
+}
+
+func (c *loopChannel) Close() {
+}
+
+func (c *loopChannel) String() string {
+	return "loop"
+}
+
+func (s *loopSocket) Open(env.Env) (Channel, error) {
+	return &loopChannel{queue.New()}, nil
+}
+
+func (s *loopSocket) String() string {
+	return "loop"
+}
+
+func (s *loopSocket) Close() {
+}
+
+func newLoopSocket() (S, error) {
+	return &loopSocket{}, nil
+}
+
 func New(desc string, env env.Env) (S, error) {
 	base, opts := opts.Parse(desc)
 	args := strings.SplitN(base, "/", 2)
@@ -152,6 +189,10 @@ func New(desc string, env env.Env) (S, error) {
 		proto, addr = args[0], args[1]
 	}
 
+	if addr == "loop" {
+		return newLoopSocket()
+	}
+
 	if proto == "" {
 		proto = "tcp"
 	}