handler muxer
Juan Batiz-Benet committed
Dec 16, 2014 at 02:56 UTC
9c11970554270327dfc44e041895bdb2a771174b
2 files changed
+170
net/mux2/mux.go
new
+105
@@ -0,0 +1,105 @@
1
+package mux
2
+
3
+import (
4
+ "errors"
5
+ "fmt"
6
+ "io"
7
+
8
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
9
+ swarm "github.com/jbenet/go-ipfs/net/swarm2"
10
+ eventlog "github.com/jbenet/go-ipfs/util/eventlog"
11
+ lgbl "github.com/jbenet/go-ipfs/util/eventlog/loggables"
12
+)
13
+
14
+var log = eventlog.Logger("mux2")
15
+
16
+// Mux provides simple stream multixplexing.
17
+// It helps you precisely when:
18
+// * You have many streams
19
+// * You have function handlers
20
+//
21
+// We use a totally ad-hoc encoding:
22
+//
23
+// <1 byte length in bytes><string name>
24
+//
25
+// So "bitswap" is 0x0762697473776170
26
+//
27
+// NOTE: only the dialer specifies this muxing line.
28
+// This is because we're using Streams :)
29
+//
30
+// WARNING: this datastructure IS NOT threadsafe.
31
+// do not modify it once it's begun serving.
32
+type Mux struct {
33
+ Default StreamHandler
34
+ Handlers map[string]StreamHandler
35
+}
36
+
37
+type StreamHandler func(s *swarm.Stream)
38
+
39
+// NextName reads the stream and returns the next protocol name
40
+// according to the muxer encoding.
41
+func (m *Mux) NextName(s io.Reader) (string, error) {
42
+
43
+ // c-string identifier
44
+ // the first byte is our length
45
+ l := make([]byte, 1)
46
+ if _, err := io.ReadFull(s, l); err != nil {
47
+ return "", err
48
+ }
49
+ length := int(l[0])
50
+
51
+ // the next are our identifier
52
+ name := make([]byte, length)
53
+ if _, err := io.ReadFull(s, name); err != nil {
54
+ return "", err
55
+ }
56
+
57
+ return string(name), nil
58
+}
59
+
60
+// NextHandler reads the stream and returns the next Handler function
61
+// according to the muxer encoding.
62
+func (m *Mux) NextHandler(s io.Reader) (string, StreamHandler, error) {
63
+ name, err := m.NextName(s)
64
+ if err != nil {
65
+ return "", nil, err
66
+ }
67
+
68
+ h, found := m.Handlers[name]
69
+ if !found {
70
+ if m.Default == nil {
71
+ return name, nil, errors.New("no handler with name: " + name)
72
+ }
73
+
74
+ return name, m.Default, nil
75
+ }
76
+
77
+ return name, h, nil
78
+}
79
+
80
+// Handle reads the next name off the Stream, and calls a function
81
+func (m *Mux) Handle(s *swarm.Stream) {
82
+ ctx := context.Background()
83
+
84
+ name, handler, err := m.NextHandler(s)
85
+ if err != nil {
86
+ err = fmt.Errorf("protocol mux error: %s", err)
87
+ log.Error(err)
88
+ log.Event(ctx, "muxError", lgbl.Error(err))
89
+ return
90
+ }
91
+
92
+ log.Info("muxer handle protocol: %s", name)
93
+ log.Event(ctx, "muxHandle", eventlog.Metadata{"protocol": name})
94
+ handler(s)
95
+}
96
+
97
+// Write writes the name into Writer with a length-byte-prefix.
98
+func Write(w io.Writer, name string) error {
99
+ s := make([]byte, len(name)+1)
100
+ s[0] = byte(len(name))
101
+ copy(s[1:], []byte(name))
102
+
103
+ _, err := w.Write(s)
104
+ return err
105
+}
net/mux2/mux_test.go
new
+65
@@ -0,0 +1,65 @@
1
+package mux
2
+
3
+import (
4
+ "bytes"
5
+ "testing"
6
+
7
+ swarm "github.com/jbenet/go-ipfs/net/swarm2"
8
+)
9
+
10
+var testCases = map[string]string{
11
+ "bitswap": "\u0007bitswap",
12
+ "dht": "\u0003dht",
13
+ "ipfs": "\u0004ipfs",
14
+ "ipfsdksnafkasnfkdajfkdajfdsjadosiaaodjasofdias": ".ipfsdksnafkasnfkdajfkdajfdsjadosiaaodjasofdias",
15
+}
16
+
17
+func TestWrite(t *testing.T) {
18
+ for k, v := range testCases {
19
+ var buf bytes.Buffer
20
+ Write(&buf, k)
21
+
22
+ v2 := buf.Bytes()
23
+ if !bytes.Equal(v2, []byte(v)) {
24
+ t.Errorf("failed: %s - %v != %v", k, []byte(v), v2)
25
+ }
26
+ }
27
+}
28
+
29
+func TestHandler(t *testing.T) {
30
+
31
+ outs := make(chan string, 10)
32
+
33
+ h := func(n string) func(s *swarm.Stream) {
34
+ return func(s *swarm.Stream) {
35
+ outs <- n
36
+ }
37
+ }
38
+
39
+ m := Mux{Handlers: map[string]StreamHandler{}}
40
+ m.Default = h("default")
41
+ m.Handlers["dht"] = h("bitswap")
42
+ // m.Handlers["ipfs"] = h("bitswap") // default!
43
+ m.Handlers["bitswap"] = h("bitswap")
44
+ m.Handlers["ipfsdksnafkasnfkdajfkdajfdsjadosiaaodjasofdias"] = h("bitswap")
45
+
46
+ for k, v := range testCases {
47
+ var buf bytes.Buffer
48
+ if _, err := buf.Write([]byte(v)); err != nil {
49
+ t.Error(err)
50
+ continue
51
+ }
52
+
53
+ name, _, err := m.NextHandler(&buf)
54
+ if err != nil {
55
+ t.Error(err)
56
+ continue
57
+ }
58
+
59
+ if name != k {
60
+ t.Errorf("name mismatch: %s != %s", k, name)
61
+ continue
62
+ }
63
+ }
64
+
65
+}