provider, node, net, io, sdk/node, examples/go/broadcast, examples/go/clock: wrap around io.PipeReader and io.PipeWriter to emit io.EOF instead of io.ErrClosedPipe, remove BindFunc in favor of strings representative of tcp addresses to listen to peers from, update readme with change in go sdk, add ProvidersFor(services ...string) to grab all available providers for a set of services in Go and NodeJS, add chatroom broadcast example for chatting across nodes using ProvidersFor()

Author: lithdew

README.md

@@ -285,13 +285,9 @@ type Node struct {
     // by calling `flatend.GenerateSecretKey()`.
     SecretKey kademlia.PrivateKey
 
-    // A list of addresses and ports assembled using:
-    // 1. flatend.BindAny() (bind to all hosts and any available port)
-    // 2. flatend.BindTCP(string) (binds to a [host]:[port])
-    // 3. flatend.BindTCPv4(string) (binds to an [IPv4 host]:[port])
-    // 4. flatend.BindTCPv6(string) (binds to an [IPv6 host]:[port])
-    // which your Flatend node will listen for other nodes from.
-    BindAddrs []BindFunc
+	// A list of IPv4/IPv6 addresses and ports assembled as [host]:[port] which
+	// your Flatend node will listen for other nodes from.
+    BindAddrs []string
 
     // A mapping of service names to their respective handlers.
     Services map[string]Handler

examples/go/broadcast/README.md

@@ -0,0 +1,100 @@
+# broadcast
+
+A demo of broadcasting a message from one node to several nodes providing the service named `'chat'`.
+
+```
+[terminal 1] $ go run main.go -l :9000
+2020/06/27 02:37:52 Listening for Flatend nodes on '[::]:9000'.
+2020/06/27 02:37:53 0.0.0.0:9001 has connected. Services: [chat]
+2020/06/27 02:37:56 0.0.0.0:9002 has connected. Services: [chat]
+hello
+Got 'world' from 0.0.0.0:9001!
+Got 'test' from 0.0.0.0:9002!
+2020/06/27 02:39:09 0.0.0.0:9002 has disconnected from you. Services: [chat]
+2020/06/27 02:39:10 0.0.0.0:9001 has disconnected from you. Services: [chat]
+
+[terminal 2] $ go run main.go -l :9001 :9000
+2020/06/27 02:37:53 Listening for Flatend nodes on '[::]:9001'.
+2020/06/27 02:37:53 You are now connected to 0.0.0.0:9000. Services: [chat]
+2020/06/27 02:37:53 Re-probed 0.0.0.0:9000. Services: [chat]
+2020/06/27 02:37:53 Discovered 0 peer(s).
+2020/06/27 02:37:56 0.0.0.0:9002 has connected. Services: [chat]
+Got 'hello' from 0.0.0.0:9000!
+world
+Got 'test' from 0.0.0.0:9002!
+2020/06/27 02:39:09 0.0.0.0:9002 has disconnected from you. Services: [chat]
+
+[terminal 3] $ go run main.go -l :9002 :9000
+2020/06/27 02:37:56 Listening for Flatend nodes on '[::]:9002'.
+2020/06/27 02:37:56 You are now connected to 0.0.0.0:9000. Services: [chat]
+2020/06/27 02:37:56 Re-probed 0.0.0.0:9000. Services: [chat]
+2020/06/27 02:37:56 You are now connected to 0.0.0.0:9001. Services: [chat]
+2020/06/27 02:37:56 Discovered 1 peer(s).
+Got 'hello' from 0.0.0.0:9000!
+Got 'world' from 0.0.0.0:9001!
+```
+
+```go
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"flag"
+	"fmt"
+	"github.com/lithdew/flatend"
+	"io/ioutil"
+	"os"
+)
+
+func check(err error) {
+	if err != nil {
+		panic(err)
+	}
+}
+
+func main() {
+	var listenAddr string
+	flag.StringVar(&listenAddr, "l", ":9000", "address to listen for peers on")
+	flag.Parse()
+
+	node := &flatend.Node{
+		PublicAddr: listenAddr,
+		BindAddrs:  []string{listenAddr},
+		SecretKey:  flatend.GenerateSecretKey(),
+		Services: map[string]flatend.Handler{
+			"chat": func(ctx *flatend.Context) {
+				buf, err := ioutil.ReadAll(ctx.Body)
+				if err != nil {
+					return
+				}
+				fmt.Printf("Got '%s' from %s:%d!\n", string(buf), ctx.ID.Host.String(), ctx.ID.Port)
+			},
+		},
+	}
+	defer node.Shutdown()
+
+	check(node.Start(flag.Args()...))
+
+	br := bufio.NewReader(os.Stdin)
+	for {
+		line, _, err := br.ReadLine()
+		if err != nil {
+			break
+		}
+
+		line = bytes.TrimSpace(line)
+		if len(line) == 0 {
+			continue
+		}
+
+		providers := node.ProvidersFor("chat")
+		for _, provider := range providers {
+			_, err := provider.Push([]string{"chat"}, nil, ioutil.NopCloser(bytes.NewReader(line)))
+			if err != nil {
+				fmt.Printf("Unable to broadcast to %s: %s\n", provider.Addr(), err)
+			}
+		}
+	}
+}
+```

examples/go/broadcast/main.go

@@ -0,0 +1,62 @@
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"flag"
+	"fmt"
+	"github.com/lithdew/flatend"
+	"io/ioutil"
+	"os"
+)
+
+func check(err error) {
+	if err != nil {
+		panic(err)
+	}
+}
+
+func main() {
+	var listenAddr string
+	flag.StringVar(&listenAddr, "l", ":9000", "address to listen for peers on")
+	flag.Parse()
+
+	node := &flatend.Node{
+		PublicAddr: listenAddr,
+		BindAddrs:  []string{listenAddr},
+		SecretKey:  flatend.GenerateSecretKey(),
+		Services: map[string]flatend.Handler{
+			"chat": func(ctx *flatend.Context) {
+				buf, err := ioutil.ReadAll(ctx.Body)
+				if err != nil {
+					return
+				}
+				fmt.Printf("Got '%s' from %s:%d!\n", string(buf), ctx.ID.Host.String(), ctx.ID.Port)
+			},
+		},
+	}
+	defer node.Shutdown()
+
+	check(node.Start(flag.Args()...))
+
+	br := bufio.NewReader(os.Stdin)
+	for {
+		line, _, err := br.ReadLine()
+		if err != nil {
+			break
+		}
+
+		line = bytes.TrimSpace(line)
+		if len(line) == 0 {
+			continue
+		}
+
+		providers := node.ProvidersFor("chat")
+		for _, provider := range providers {
+			_, err := provider.Push([]string{"chat"}, nil, ioutil.NopCloser(bytes.NewReader(line)))
+			if err != nil {
+				fmt.Printf("Unable to broadcast to %s: %s\n", provider.Addr(), err)
+			}
+		}
+	}
+}

examples/go/clock/README.md

@@ -83,7 +83,7 @@ func clock(ctx *flatend.Context) {
 	ours := latest.Format(time.Stamp)
 
 	timestamp, err := ioutil.ReadAll(ctx.Body)
-	if err != nil && !errors.Is(err, io.ErrClosedPipe) {
+	if err != nil {
 		return
 	}
 
@@ -136,9 +136,7 @@ func main() {
 				fmt.Printf("[%d] Asked someone for their current time. Ours is '%s'.\n", i, timestamp)
 
 				res, err := ioutil.ReadAll(stream.Reader)
-				if !errors.Is(err, io.ErrClosedPipe) {
-					check(err)
-				}
+				check(err)
 
 				fmt.Printf("[%d] Got a response! Their current time is: '%s'\n", i, string(res))
 			}

examples/go/clock/main.go

@@ -25,7 +25,7 @@ func clock(ctx *flatend.Context) {
 	ours := latest.Format(time.Stamp)
 
 	timestamp, err := ioutil.ReadAll(ctx.Body)
-	if err != nil && !errors.Is(err, io.ErrClosedPipe) {
+	if err != nil {
 		return
 	}
 

io.go

@@ -0,0 +1,41 @@
+package flatend
+
+import (
+	"errors"
+	"fmt"
+	"io"
+)
+
+type pipeReader struct {
+	*io.PipeReader
+}
+
+func (r *pipeReader) Read(buf []byte) (n int, err error) {
+	n, err = r.PipeReader.Read(buf)
+	if err != nil && errors.Is(err, io.ErrClosedPipe) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+type pipeWriter struct {
+	*io.PipeWriter
+}
+
+func (w *pipeWriter) Write(buf []byte) (n int, err error) {
+	n, err = w.PipeWriter.Write(buf)
+	if err != nil && errors.Is(err, io.ErrClosedPipe) {
+		err = fmt.Errorf("%s: %w", err, io.EOF)
+	}
+	return n, err
+}
+
+// createWrappedPipe wraps around a reader/writer pair from io.Pipe() such that all
+// errors reported by such reader/writer pair that comprise of io.ErrClosedPipe
+// will be wrapped with io.EOF.
+func createWrappedPipe() (*pipeReader, *pipeWriter) {
+	r, w := io.Pipe()
+	pr := &pipeReader{PipeReader: r}
+	pw := &pipeWriter{PipeWriter: w}
+	return pr, pw
+}

net.go

@@ -4,30 +4,11 @@ import (
 	"github.com/lithdew/kademlia"
 	"github.com/lithdew/monte"
 	"io"
-	"net"
 	"sync"
 )
 
 const ChunkSize = 2048
 
-type BindFunc func() (net.Listener, error)
-
-func BindAny() BindFunc {
-	return func() (net.Listener, error) { return net.Listen("tcp", ":0") }
-}
-
-func BindTCP(addr string) BindFunc {
-	return func() (net.Listener, error) { return net.Listen("tcp", addr) }
-}
-
-func BindTCPv4(addr string) BindFunc {
-	return func() (net.Listener, error) { return net.Listen("tcp4", addr) }
-}
-
-func BindTCPv6(addr string) BindFunc {
-	return func() (net.Listener, error) { return net.Listen("tcp6", addr) }
-}
-
 var _ io.Writer = (*Context)(nil)
 
 type Context struct {