stream-server/browser.js at master · Colingo/stream-server · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
var hat = require("hat")
    , rack = hat.rack(128, 16, 16)
    , through = require("through")
    , PauseStream = require("pause-stream")
    , duplex = require("duplexer")

module.exports = createServer

function createServer(mdm, options, callback) {
    if (typeof options === "function") {
        callback = options
        options = {}
    }

    options = options || {}
    var prefix = options.prefix || "/stream-server"

    return {
        listen: listen
        , connect: connect
    }

    function connect(name) {
        // Return a proxy stream
        var readable = through(catchFirstMessage)
            // catch first message from server that stream is open
            , writable = PauseStream().pause()
            , proxy = duplex(writable, readable)
            , tryAgain = 4
            , connected = false

        // try to open the stream
        openStream()

        return proxy

        function openStream() {
            // connect to a server directly
            var id = rack()
                , stream = mdm.createStream(prefix + "/client/" + name +
                    "/" + id)

            writable.pipe(stream).pipe(readable)

            stream.once("ended", function () {
                stream.end()
            })

            // handle server does not exist yet and try to reconnect 4 times
            stream.once("error", handleError)

            function handleError(err) {
                var code = err.code
                    , message = err.message

                if (code === "404" && message === "server does not exist") {
                    tryAgain--
                    if (tryAgain === 0) {
                        return proxy.emit("error", err)
                    }

                    stream.end()
                    stream.destroy()
                    setTimeout(openStream, 500)
                } else {
                    proxy.emit("error", err)
                }
            }
        }

        function catchFirstMessage(data) {
            // the open message is from the proxy server
            if (connected === false && data === "[stream-server]:open") {
                connected = true
                writable.resume()
            } else {
                this.emit("data", data)
            }
        }
    }

    function listen(name) {
        // open a server stream
        var stream = mdm.createStream(prefix + "/server/" + name)

        stream.on("data", openServerConnection)

        return stream

        function openServerConnection(clientName) {
            // for each client message that comes up create a proxy stream
            // and return it
            var writable = PauseStream().pause()
                , readable = through(catchFirstMessage)
                , proxy = duplex(writable, readable)
                , tryAgain = 4
                , connected = false

            openStream()

            callback(proxy)

            function openStream() {
                // connect to client directly (through relay)
                var id = rack()
                    , stream = mdm.createStream(prefix + "/server/" + name +
                    "/client/" + clientName + "/" + id)

                // Boot server says the relay server disconnected.
                // Clean up open streams
                stream.once("ended", function () {
                    stream.purge && stream.purge()
                    stream.end()
                })

                writable.pipe(stream).pipe(readable)

                // handle client does not exist yet and try to reconnect
                stream.once("error", handleError)

                function handleError(err) {
                    var code = err.code
                        , message = err.message

                    if (code === "500" && message === "cannot open client-" +
                        "server connection"
                    ) {
                        tryAgain--
                        if (tryAgain === 0) {
                            return proxy.emit("error", err)
                        }

                        stream.end()
                        stream.destroy()
                        setTimeout(openStream, 500)
                    } else {
                        proxy.emit("error", err)
                    }
                }
            }

            function catchFirstMessage(data) {
                // the open message is from the proxy server
                if (connected === false && data === "[stream-server]:open") {
                    connected = true
                    writable.resume()
                } else {
                    this.emit("data", data)
                }
            }
        }
    }
}