Implementation

Makefile

@@ -0,0 +1,10 @@
+bin: deps
+	go install ./...
+
+gx:
+	go get github.com/whyrusleeping/gx
+	go get github.com/whyrusleeping/gx-go
+
+deps: gx
+	gx --verbose install --global
+	gx-go rewrite

conn.go

@@ -0,0 +1,190 @@
+package p2pd
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net"
+	"time"
+
+	pb "github.com/libp2p/go-libp2p-daemon/pb"
+
+	ggio "github.com/gogo/protobuf/io"
+	inet "github.com/libp2p/go-libp2p-net"
+	peer "github.com/libp2p/go-libp2p-peer"
+	pstore "github.com/libp2p/go-libp2p-peerstore"
+	proto "github.com/libp2p/go-libp2p-protocol"
+	ma "github.com/multiformats/go-multiaddr"
+)
+
+const DefaultTimeout = 60 * time.Second
+
+func (d *Daemon) handleConn(c net.Conn) {
+	defer c.Close()
+
+	r := ggio.NewDelimitedReader(c, inet.MessageSizeMax)
+	w := ggio.NewDelimitedWriter(c)
+
+	for {
+		var req pb.Request
+
+		err := r.ReadMsg(&req)
+		if err != nil {
+			if err != io.EOF {
+				log.Debugf("Error reading message: %s", err.Error())
+			}
+			return
+		}
+
+		switch *req.Type {
+		case pb.Request_CONNECT:
+			res := d.doConnect(&req)
+			err := w.WriteMsg(res)
+			if err != nil {
+				log.Debugf("Error writing response: %s", err.Error())
+				return
+			}
+
+		case pb.Request_STREAM_OPEN:
+			res, s := d.doStreamOpen(&req)
+			err := w.WriteMsg(res)
+			if err != nil {
+				log.Debugf("Error writing response: %s", err.Error())
+				if s != nil {
+					s.Reset()
+				}
+				return
+			}
+
+			if s != nil {
+				d.doStreamPipe(c, s)
+				return
+			}
+
+		case pb.Request_STREAM_HANDLER:
+			res := d.doStreamHandler(&req)
+			err := w.WriteMsg(res)
+			if err != nil {
+				log.Debugf("Error writing response: %s", err.Error())
+				return
+			}
+
+		default:
+			log.Debugf("Unexpected request type: %s", req.Type)
+			return
+		}
+	}
+}
+
+func (d *Daemon) doConnect(req *pb.Request) *pb.Response {
+	ctx, cancel := context.WithTimeout(d.ctx, DefaultTimeout)
+	defer cancel()
+
+	if req.Connect == nil {
+		return errorResponse(errors.New("Malformed request; missing parameters"))
+	}
+
+	pid, err := peer.IDFromBytes(req.Connect.Peer)
+	if err != nil {
+		log.Debugf("Error parsing peer ID: %s", err.Error())
+		return errorResponse(err)
+	}
+
+	var addrs []ma.Multiaddr
+	addrs = make([]ma.Multiaddr, len(req.Connect.Addrs))
+	for x, bs := range req.Connect.Addrs {
+		addr, err := ma.NewMultiaddrBytes(bs)
+		if err != nil {
+			log.Debugf("Error parsing multiaddr: %s", err.Error())
+			return errorResponse(err)
+		}
+		addrs[x] = addr
+	}
+
+	pi := pstore.PeerInfo{ID: pid, Addrs: addrs}
+
+	log.Debugf("connecting to %s", pid.Pretty())
+	err = d.host.Connect(ctx, pi)
+	if err != nil {
+		log.Debugf("error opening connection to %s: %s", pid.Pretty(), err.Error())
+		return errorResponse(err)
+	}
+
+	return okResponse()
+}
+
+func (d *Daemon) doStreamOpen(req *pb.Request) (*pb.Response, inet.Stream) {
+	ctx, cancel := context.WithTimeout(d.ctx, DefaultTimeout)
+	defer cancel()
+
+	if req.StreamOpen == nil {
+		return errorResponse(errors.New("Malformed request; missing parameters")), nil
+	}
+
+	pid, err := peer.IDFromBytes(req.StreamOpen.Peer)
+	if err != nil {
+		log.Debugf("Error parsing peer ID: %s", err.Error())
+		return errorResponse(err), nil
+	}
+
+	protos := make([]proto.ID, len(req.StreamOpen.Proto))
+	for x, str := range req.StreamOpen.Proto {
+		protos[x] = proto.ID(str)
+	}
+
+	log.Debugf("opening stream to %s", pid.Pretty())
+	s, err := d.host.NewStream(ctx, pid, protos...)
+	if err != nil {
+		log.Debugf("Error opening stream to %s: %s", pid.Pretty(), err.Error())
+		return errorResponse(err), nil
+	}
+
+	res := okResponse()
+	res.StreamInfo = makeStreamInfo(s)
+	return res, s
+}
+
+func (d *Daemon) doStreamHandler(req *pb.Request) *pb.Response {
+	if req.StreamHandler == nil {
+		return errorResponse(errors.New("Malformed request; missing parameters"))
+	}
+
+	d.mx.Lock()
+	defer d.mx.Unlock()
+
+	path := *req.StreamHandler.Path
+	for sp := range req.StreamHandler.Proto {
+		p := proto.ID(sp)
+		_, ok := d.handlers[p]
+		if !ok {
+			d.host.SetStreamHandler(p, d.handleStream)
+		}
+		log.Debugf("set stream handler: %s -> %s", p, path)
+		d.handlers[p] = path
+	}
+
+	return okResponse()
+}
+
+func okResponse() *pb.Response {
+	return &pb.Response{
+		Type: pb.Response_OK.Enum(),
+	}
+}
+
+func errorResponse(err error) *pb.Response {
+	errstr := err.Error()
+	return &pb.Response{
+		Type:  pb.Response_ERROR.Enum(),
+		Error: &pb.ErrorResponse{Msg: &errstr},
+	}
+}
+
+func makeStreamInfo(s inet.Stream) *pb.StreamInfo {
+	proto := string(s.Protocol())
+	return &pb.StreamInfo{
+		Peer:  []byte(s.Conn().RemotePeer()),
+		Addr:  s.Conn().RemoteMultiaddr().Bytes(),
+		Proto: &proto,
+	}
+}

daemon.go

@@ -0,0 +1,60 @@
+package p2pd
+
+import (
+	"context"
+	"net"
+	"sync"
+
+	logging "github.com/ipfs/go-log"
+	libp2p "github.com/libp2p/go-libp2p"
+	host "github.com/libp2p/go-libp2p-host"
+	proto "github.com/libp2p/go-libp2p-protocol"
+)
+
+var log = logging.Logger("p2pd")
+
+type Daemon struct {
+	ctx      context.Context
+	host     host.Host
+	listener net.Listener
+
+	mx sync.Mutex
+	// stream handlers: map of protocol.ID to unix socket path
+	handlers map[proto.ID]string
+}
+
+func NewDaemon(ctx context.Context, path string, opts ...libp2p.Option) (*Daemon, error) {
+	h, err := libp2p.New(ctx, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	l, err := net.Listen("unix", path)
+	if err != nil {
+		h.Close()
+		return nil, err
+	}
+
+	d := &Daemon{
+		ctx:      ctx,
+		host:     h,
+		listener: l,
+		handlers: make(map[proto.ID]string),
+	}
+
+	go d.listen()
+
+	return d, nil
+}
+
+func (d *Daemon) listen() {
+	for {
+		c, err := d.listener.Accept()
+		if err != nil {
+			log.Errorf("error accepting connection: %s", err.Error())
+		}
+
+		log.Debug("incoming connection")
+		go d.handleConn(c)
+	}
+}

p2pd/main.go

@@ -0,0 +1,17 @@
+package main
+
+import (
+	"context"
+	"log"
+
+	p2pd "github.com/libp2p/go-libp2p-daemon"
+)
+
+func main() {
+	_, err := p2pd.NewDaemon(context.Background(), "/tmp/p2pd.sock")
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	select {}
+}

package.json

@@ -0,0 +1,69 @@
+{
+  "author": "vyzo",
+  "bugs": {},
+  "gx": {
+    "dvcsimport": "github.com/libp2p/go-libp2p-daemon"
+  },
+  "gxDependencies": [
+    {
+      "author": "whyrusleeping",
+      "hash": "QmUEqyXr97aUbNmQADHYNknjwjjdVpJXEt1UZXmSG81EV4",
+      "name": "go-libp2p",
+      "version": "6.0.12"
+    },
+    {
+      "author": "whyrusleeping",
+      "hash": "QmeMYW7Nj8jnnEfs9qhm7SxKkoDPUWXu3MsxX6BFwz34tf",
+      "name": "go-libp2p-host",
+      "version": "3.0.9"
+    },
+    {
+      "author": "whyrusleeping",
+      "hash": "QmZNkThpqfVXs9GNbexPrfBbXSLNYeKrE7jwFM2oqHbyqN",
+      "name": "go-libp2p-protocol",
+      "version": "1.0.0"
+    },
+    {
+      "hash": "Qmbq7kGxgcpALGLPaWDyTa6KUq5kBUKdEvkvPZcBkJoLex",
+      "name": "go-log",
+      "version": "1.5.6"
+    },
+    {
+      "author": "whyrusleeping",
+      "hash": "QmQsErDt8Qgw1XrsXf2BpEzDgGWtB1YLsTAARBup5b6B9W",
+      "name": "go-libp2p-peer",
+      "version": "2.3.7"
+    },
+    {
+      "author": "whyrusleeping",
+      "hash": "QmZNJyx9GGCX4GeuHnLB8fxaxMLs4MjTjHokxfQcCd6Nve",
+      "name": "go-libp2p-net",
+      "version": "3.0.9"
+    },
+    {
+      "author": "multiformats",
+      "hash": "QmYmsdtJ3HsodkePE3eU3TsCaP2YvPZJ4LoXnNkDE5Tpt7",
+      "name": "go-multiaddr",
+      "version": "1.3.0"
+    },
+    {
+      "author": "whyrusleeping",
+      "hash": "QmdxUuburamoF6zF9qjeQC4WYcWGbWuRmdLacMEsW8ioD8",
+      "name": "gogo-protobuf",
+      "version": "0.0.0"
+    },
+    {
+      "author": "whyrusleeping",
+      "hash": "Qmda4cPRvSRyox3SqgJN6DfSZGU5TtHufPTp9uXjFj71X6",
+      "name": "go-libp2p-peerstore",
+      "version": "2.0.0"
+    }
+  ],
+  "gxVersion": "0.12.1",
+  "language": "go",
+  "license": "",
+  "name": "go-libp2p-daemon",
+  "releaseCmd": "git commit -a -m \"gx publish $VERSION\"",
+  "version": "0.0.0"
+}
+