Compare commits
38 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 James Coglan
|
ff2a854ead | ||
|
James Coglan
|
7ad3317a26 | ||
|
James Coglan
|
0ae6ad9a9d | ||
|
James Coglan
|
5ea0b42080 | ||
|
James Coglan
|
29496f6838 | ||
|
James Coglan
|
4a76c75efb | ||
|
James Coglan
|
44a677a9c0 | ||
|
James Coglan
|
f6c50aba0c | ||
|
James Coglan
|
2d211f3705 | ||
|
James Coglan
|
0b620834cc | ||
|
James Coglan
|
729a465307 | ||
|
James Coglan
|
2af2c18251 | ||
|
James Coglan
|
50bcedda78 | ||
|
James Coglan
|
e3aa5246d6 | ||
|
James Coglan
|
1e58c148cb | ||
|
James Coglan
|
5f040a15af | ||
|
James Coglan
|
654d9b0acc | ||
|
James Coglan
|
c37d0611c7 | ||
|
James Coglan
|
d8d38e54e6 | ||
|
James Coglan
|
fb84d36546 | ||
|
James Coglan
|
36cc2c5c73 | ||
|
James Coglan
|
b315aa08d6 | ||
|
James Coglan
|
b23eb5b890 | ||
|
James Coglan
|
7319766a5e | ||
|
James Coglan
|
9418affa01 | ||
|
James Coglan
|
b27d4cebf8 | ||
|
James Coglan
|
2792339b4d | ||
|
James Coglan
|
5e5f1f454f | ||
|
James Coglan
|
2365c0aef2 | ||
|
James Coglan
|
6669e323c3 | ||
|
James Coglan
|
c8f31cc1c7 | ||
|
James Coglan
|
62ac506b80 | ||
|
James Coglan
|
89104ddd48 | ||
|
James Coglan
|
2a39f8eaac | ||
|
James Coglan
|
eb6718f6c8 | ||
|
James Coglan
|
a04471aecc | ||
|
James Coglan
|
6a9e862c07 | ||
|
James Coglan
|
1b77290e82 |
@@ -0,0 +1,41 @@
|
||||
on:
|
||||
- push
|
||||
- pull_request
|
||||
|
||||
jobs:
|
||||
test:
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
node:
|
||||
- '0.8'
|
||||
- '0.10'
|
||||
- '0.12'
|
||||
- '4'
|
||||
- '6'
|
||||
- '8'
|
||||
- '10'
|
||||
- '12'
|
||||
- '14'
|
||||
- '16'
|
||||
- '18'
|
||||
- '20'
|
||||
name: node.js v${{ matrix.node }}
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v2
|
||||
|
||||
- uses: actions/setup-node@v2
|
||||
with:
|
||||
node-version: ${{ matrix.node }}
|
||||
|
||||
- if: matrix.node == '0.8'
|
||||
run: npm conf set strict-ssl false
|
||||
|
||||
- run: node --version
|
||||
- run: npm install
|
||||
|
||||
- run: npm install 'nopt@5'
|
||||
- run: rm -rf node_modules/jstest/node_modules/nopt
|
||||
|
||||
- run: npm test
|
||||
@@ -1 +1,2 @@
|
||||
node_modules
|
||||
package-lock.json
|
||||
|
||||
@@ -1,6 +0,0 @@
|
||||
.git
|
||||
.gitignore
|
||||
.npmignore
|
||||
.travis.yml
|
||||
node_modules
|
||||
spec
|
||||
-10
@@ -1,10 +0,0 @@
|
||||
language: node_js
|
||||
|
||||
node_js:
|
||||
- "0.6"
|
||||
- "0.8"
|
||||
- "0.10"
|
||||
- "0.11"
|
||||
|
||||
before_install:
|
||||
- '[ "${TRAVIS_NODE_VERSION}" = "0.6" ] && npm conf set strict-ssl false || true'
|
||||
@@ -0,0 +1,28 @@
|
||||
### 0.1.4 / 2020-06-02
|
||||
|
||||
- Remove a ReDoS vulnerability in the header parser (CVE-2020-7662, reported by
|
||||
Robert McLaughlin)
|
||||
- Change license from MIT to Apache 2.0
|
||||
|
||||
### 0.1.3 / 2017-11-11
|
||||
|
||||
- Accept extension names and parameters including uppercase letters
|
||||
- Handle extension names that clash with `Object.prototype` properties
|
||||
|
||||
### 0.1.2 / 2017-09-10
|
||||
|
||||
- Catch synchronous exceptions thrown when calling an extension
|
||||
- Fix race condition caused when a message is pushed after a cell has stopped
|
||||
due to an error
|
||||
- Fix failure of `close()` to return if a message that's queued after one that
|
||||
produces an error never finishes being processed
|
||||
|
||||
### 0.1.1 / 2015-02-19
|
||||
|
||||
- Prevent sessions being closed before they have finished processing messages
|
||||
- Add a callback to `Extensions.close()` so the caller can tell when it's safe
|
||||
to close the socket
|
||||
|
||||
### 0.1.0 / 2014-12-12
|
||||
|
||||
- Initial release
|
||||
@@ -0,0 +1,4 @@
|
||||
# Code of Conduct
|
||||
|
||||
All projects under the [Faye](https://github.com/faye) umbrella are covered by
|
||||
the [Code of Conduct](https://github.com/faye/code-of-conduct).
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
Copyright 2014-2020 James Coglan
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License"); you may not use
|
||||
this file except in compliance with the License. You may obtain a copy of the
|
||||
License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software distributed
|
||||
under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
|
||||
CONDITIONS OF ANY KIND, either express or implied. See the License for the
|
||||
specific language governing permissions and limitations under the License.
|
||||
@@ -1,4 +1,4 @@
|
||||
# websocket-extensions [](http://travis-ci.org/faye/websocket-extensions-node)
|
||||
# websocket-extensions
|
||||
|
||||
A minimal framework that supports the implementation of WebSocket extensions in
|
||||
a way that's decoupled from the main protocol. This library aims to allow a
|
||||
@@ -183,9 +183,12 @@ At the end of the WebSocket session (either when the protocol is explicitly
|
||||
ended or the transport connection disconnects), the driver should call:
|
||||
|
||||
```js
|
||||
exts.close()
|
||||
exts.close(function() {})
|
||||
```
|
||||
|
||||
The callback is invoked when all extensions have finished processing any
|
||||
messages in the pipeline and it's safe to close the socket.
|
||||
|
||||
### For extension authors
|
||||
|
||||
An extension author is someone implementing an extension that transforms
|
||||
@@ -231,8 +234,8 @@ then the `permessage-deflate` extension will receive the call:
|
||||
|
||||
```js
|
||||
ext.createServerSession([
|
||||
{server_no_context_takeover: true, server_max_window_bits: 8},
|
||||
{server_max_window_bits: 15}
|
||||
{ server_no_context_takeover: true, server_max_window_bits: 8 },
|
||||
{ server_max_window_bits: 15 }
|
||||
]);
|
||||
```
|
||||
|
||||
@@ -248,8 +251,8 @@ implement the following methods, as well as the *Session* API listed below.
|
||||
```js
|
||||
clientSession.generateOffer()
|
||||
// e.g. -> [
|
||||
// {server_no_context_takeover: true, server_max_window_bits: 8},
|
||||
// {server_max_window_bits: 15}
|
||||
// { server_no_context_takeover: true, server_max_window_bits: 8 },
|
||||
// { server_max_window_bits: 15 }
|
||||
// ]
|
||||
```
|
||||
|
||||
@@ -274,7 +277,7 @@ must implement the following methods, as well as the *Session* API listed below.
|
||||
|
||||
```js
|
||||
serverSession.generateResponse()
|
||||
// e.g. -> {server_max_window_bits: 8}
|
||||
// e.g. -> { server_max_window_bits: 8 }
|
||||
```
|
||||
|
||||
This returns the set of parameters the server session wants to send in its
|
||||
@@ -324,29 +327,5 @@ the session to release any resources it's using.
|
||||
|
||||
## Examples
|
||||
|
||||
* Consumer: [websocket-driver](https://github.com/faye/websocket-driver-node)
|
||||
* Provider: [permessage-deflate](https://github.com/faye/permessage-deflate-node)
|
||||
|
||||
## License
|
||||
|
||||
(The MIT License)
|
||||
|
||||
Copyright (c) 2014 James Coglan
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
this software and associated documentation files (the 'Software'), to deal in
|
||||
the Software without restriction, including without limitation the rights to
|
||||
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
|
||||
of the Software, and to permit persons to whom the Software is furnished to do
|
||||
so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED 'AS IS', WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE.
|
||||
- Consumer: [websocket-driver](https://github.com/faye/websocket-driver-node)
|
||||
- Provider: [permessage-deflate](https://github.com/faye/permessage-deflate-node)
|
||||
|
||||
+10
-6
@@ -1,13 +1,15 @@
|
||||
'use strict';
|
||||
|
||||
var TOKEN = /([!#\$%&'\*\+\-\.\^_`\|~0-9a-z]+)/,
|
||||
NOTOKEN = /([^!#\$%&'\*\+\-\.\^_`\|~0-9a-z])/g,
|
||||
QUOTED = /"((?:\\[\x00-\x7f]|[^\x00-\x08\x0a-\x1f\x7f"])*)"/,
|
||||
var TOKEN = /([!#\$%&'\*\+\-\.\^_`\|~0-9A-Za-z]+)/,
|
||||
NOTOKEN = /([^!#\$%&'\*\+\-\.\^_`\|~0-9A-Za-z])/g,
|
||||
QUOTED = /"((?:\\[\x00-\x7f]|[^\x00-\x08\x0a-\x1f\x7f"\\])*)"/,
|
||||
PARAM = new RegExp(TOKEN.source + '(?:=(?:' + TOKEN.source + '|' + QUOTED.source + '))?'),
|
||||
EXT = new RegExp(TOKEN.source + '(?: *; *' + PARAM.source + ')*', 'g'),
|
||||
EXT_LIST = new RegExp('^' + EXT.source + '(?: *, *' + EXT.source + ')*$'),
|
||||
NUMBER = /^-?(0|[1-9][0-9]*)(\.[0-9]+)?$/;
|
||||
|
||||
var hasOwnProperty = Object.prototype.hasOwnProperty;
|
||||
|
||||
var Parser = {
|
||||
parseHeader: function(header) {
|
||||
var offers = new Offers();
|
||||
@@ -35,7 +37,7 @@ var Parser = {
|
||||
}
|
||||
if (NUMBER.test(data)) data = parseFloat(data);
|
||||
|
||||
if (offer.hasOwnProperty(key)) {
|
||||
if (hasOwnProperty.call(offer, key)) {
|
||||
offer[key] = [].concat(offer[key]);
|
||||
offer[key].push(data);
|
||||
} else {
|
||||
@@ -77,9 +79,11 @@ var Offers = function() {
|
||||
};
|
||||
|
||||
Offers.prototype.push = function(name, params) {
|
||||
this._byName[name] = this._byName[name] || [];
|
||||
if (!hasOwnProperty.call(this._byName, name))
|
||||
this._byName[name] = [];
|
||||
|
||||
this._byName[name].push(params);
|
||||
this._inOrder.push({name: name, params: params});
|
||||
this._inOrder.push({ name: name, params: params });
|
||||
};
|
||||
|
||||
Offers.prototype.eachOffer = function(callback, context) {
|
||||
|
||||
@@ -1,66 +0,0 @@
|
||||
'use strict';
|
||||
|
||||
var bind = function(object, method) {
|
||||
return function() {
|
||||
return object[method].apply(object, arguments);
|
||||
};
|
||||
};
|
||||
|
||||
var Pipeline = function(sessions) {
|
||||
var nextIn, nextOut;
|
||||
|
||||
for (var i = 0, n = sessions.length; i < n; i++) {
|
||||
nextIn = new Queue(sessions[i], 'processIncomingMessage', nextIn);
|
||||
nextOut = new Queue(sessions[n-1-i], 'processOutgoingMessage', nextOut);
|
||||
}
|
||||
this._in = nextIn;
|
||||
this._out = nextOut;
|
||||
};
|
||||
|
||||
Pipeline.prototype.processIncomingMessage = function(message, callback, context) {
|
||||
if (this._in)
|
||||
this._in.push({message: message, callback: callback, context: context});
|
||||
else
|
||||
callback.call(context, null, message);
|
||||
};
|
||||
|
||||
Pipeline.prototype.processOutgoingMessage = function(message, callback, context) {
|
||||
if (this._out)
|
||||
this._out.push({message: message, callback: callback, context: context});
|
||||
else
|
||||
callback.call(context, null, message);
|
||||
};
|
||||
|
||||
var Queue = function(record, method, next) {
|
||||
this._ext = record[0];
|
||||
this._fn = bind(record[1], method);
|
||||
this._next = next;
|
||||
this._inbox = [];
|
||||
};
|
||||
|
||||
Queue.prototype.push = function(record) {
|
||||
var self = this;
|
||||
|
||||
record.done = false;
|
||||
this._inbox.push(record);
|
||||
|
||||
this._fn(record.message, function(error, msg) {
|
||||
if (error) {
|
||||
error.message = [self._ext.name, error.message].join(': ');
|
||||
return record.callback.call(record.context, error, null);
|
||||
}
|
||||
|
||||
record.message = msg;
|
||||
record.done = true;
|
||||
|
||||
while (self._inbox.length > 0 && self._inbox[0].done) {
|
||||
record = self._inbox.shift();
|
||||
if (self._next)
|
||||
self._next.push(record);
|
||||
else
|
||||
record.callback.call(record.context, null, record.message);
|
||||
}
|
||||
});
|
||||
};
|
||||
|
||||
module.exports = Pipeline;
|
||||
@@ -0,0 +1,607 @@
|
||||
# Extension pipelining
|
||||
|
||||
`websocket-extensions` models the extension negotiation and processing pipeline
|
||||
of the WebSocket protocol. Between the driver parsing messages from the TCP
|
||||
stream and handing those messages off to the application, there may exist a
|
||||
stack of extensions that transform the message somehow.
|
||||
|
||||
In the parlance of this framework, a *session* refers to a single instance of an
|
||||
extension, acting on a particular socket on either the server or the client
|
||||
side. A session may transform messages both incoming to the application and
|
||||
outgoing from the application, for example the `permessage-deflate` extension
|
||||
compresses outgoing messages and decompresses incoming messages. Message streams
|
||||
in either direction are independent; that is, incoming and outgoing messages
|
||||
cannot be assumed to 'pair up' as in a request-response protocol.
|
||||
|
||||
Asynchronous processing of messages poses a number of problems that this
|
||||
pipeline construction is intended to solve.
|
||||
|
||||
|
||||
## Overview
|
||||
|
||||
Logically, we have the following:
|
||||
|
||||
|
||||
+-------------+ out +---+ +---+ +---+ +--------+
|
||||
| |------>| |---->| |---->| |------>| |
|
||||
| Application | | A | | B | | C | | Driver |
|
||||
| |<------| |<----| |<----| |<------| |
|
||||
+-------------+ in +---+ +---+ +---+ +--------+
|
||||
|
||||
\ /
|
||||
+----------o----------+
|
||||
|
|
||||
sessions
|
||||
|
||||
|
||||
For outgoing messages, the driver receives the result of
|
||||
|
||||
C.outgoing(B.outgoing(A.outgoing(message)))
|
||||
|
||||
or, [A, B, C].reduce(((m, ext) => ext.outgoing(m)), message)
|
||||
|
||||
For incoming messages, the application receives the result of
|
||||
|
||||
A.incoming(B.incoming(C.incoming(message)))
|
||||
|
||||
or, [C, B, A].reduce(((m, ext) => ext.incoming(m)), message)
|
||||
|
||||
A session is of the following type, to borrow notation from pseudo-Haskell:
|
||||
|
||||
type Session = {
|
||||
incoming :: Message -> Message
|
||||
outgoing :: Message -> Message
|
||||
close :: () -> ()
|
||||
}
|
||||
|
||||
(That `() -> ()` syntax is intended to mean that `close()` is a nullary void
|
||||
method; I apologise to any Haskell readers for not using the right monad.)
|
||||
|
||||
The `incoming()` and `outgoing()` methods perform message transformation in the
|
||||
respective directions; `close()` is called when a socket closes so the session
|
||||
can release any resources it's holding, for example a DEFLATE de/compression
|
||||
context.
|
||||
|
||||
However because this is JavaScript, the `incoming()` and `outgoing()` methods
|
||||
may be asynchronous (indeed, `permessage-deflate` is based on `zlib`, whose API
|
||||
is stream-based). So their interface is strictly:
|
||||
|
||||
type Session = {
|
||||
incoming :: Message -> Callback -> ()
|
||||
outgoing :: Message -> Callback -> ()
|
||||
close :: () -> ()
|
||||
}
|
||||
|
||||
type Callback = Either Error Message -> ()
|
||||
|
||||
This means a message *m2* can be pushed into a session while it's still
|
||||
processing the preceding message *m1*. The messages can be processed
|
||||
concurrently but they *must* be given to the next session in line (or to the
|
||||
application) in the same order they came in. Applications will expect to receive
|
||||
messages in the order they arrived over the wire, and sessions require this too.
|
||||
So ordering of messages must be preserved throughout the pipeline.
|
||||
|
||||
Consider the following highly simplified extension that deflates messages on the
|
||||
wire. `message` is a value conforming the type:
|
||||
|
||||
type Message = {
|
||||
rsv1 :: Boolean
|
||||
rsv2 :: Boolean
|
||||
rsv3 :: Boolean
|
||||
opcode :: Number
|
||||
data :: Buffer
|
||||
}
|
||||
|
||||
Here's the extension:
|
||||
|
||||
```js
|
||||
var zlib = require('zlib');
|
||||
|
||||
var deflate = {
|
||||
outgoing: function(message, callback) {
|
||||
zlib.deflateRaw(message.data, function(error, result) {
|
||||
message.rsv1 = true;
|
||||
message.data = result;
|
||||
callback(error, message);
|
||||
});
|
||||
},
|
||||
|
||||
incoming: function(message, callback) {
|
||||
// decompress inbound messages (elided)
|
||||
},
|
||||
|
||||
close: function() {
|
||||
// no state to clean up
|
||||
}
|
||||
};
|
||||
```
|
||||
|
||||
We can call it with a large message followed by a small one, and the small one
|
||||
will be returned first:
|
||||
|
||||
```js
|
||||
var crypto = require('crypto'),
|
||||
large = crypto.randomBytes(1 << 14),
|
||||
small = new Buffer('hi');
|
||||
|
||||
deflate.outgoing({ data: large }, function() {
|
||||
console.log(1, 'large');
|
||||
});
|
||||
|
||||
deflate.outgoing({ data: small }, function() {
|
||||
console.log(2, 'small');
|
||||
});
|
||||
|
||||
/* prints: 2 'small'
|
||||
1 'large' */
|
||||
```
|
||||
|
||||
So a session that processes messages asynchronously may fail to preserve message
|
||||
ordering.
|
||||
|
||||
Now, this extension is stateless, so it can process messages in any order and
|
||||
still produce the same output. But some extensions are stateful and require
|
||||
message order to be preserved.
|
||||
|
||||
For example, when using `permessage-deflate` without `no_context_takeover` set,
|
||||
the session retains a DEFLATE de/compression context between messages, which
|
||||
accumulates state as it consumes data (later messages can refer to sections of
|
||||
previous ones to improve compression). Reordering parts of the DEFLATE stream
|
||||
will result in a failed decompression. Messages must be decompressed in the same
|
||||
order they were compressed by the peer in order for the DEFLATE protocol to
|
||||
work.
|
||||
|
||||
Finally, there is the problem of closing a socket. When a WebSocket is closed by
|
||||
the application, or receives a closing request from the other peer, there may be
|
||||
messages outgoing from the application and incoming from the peer in the
|
||||
pipeline. If we close the socket and pipeline immediately, two problems arise:
|
||||
|
||||
* We may send our own closing frame to the peer before all prior messages we
|
||||
sent have been written to the socket, and before we have finished processing
|
||||
all prior messages from the peer
|
||||
* The session may be instructed to close its resources (e.g. its de/compression
|
||||
context) while it's in the middle of processing a message, or before it has
|
||||
received messages that are upstream of it in the pipeline
|
||||
|
||||
Essentially, we must defer closing the sessions and sending a closing frame
|
||||
until after all prior messages have exited the pipeline.
|
||||
|
||||
|
||||
## Design goals
|
||||
|
||||
* Message order must be preserved between the protocol driver, the extension
|
||||
sessions, and the application
|
||||
* Messages should be handed off to sessions and endpoints as soon as possible,
|
||||
to maximise throughput of stateless sessions
|
||||
* The closing procedure should block any further messages from entering the
|
||||
pipeline, and should allow all existing messages to drain
|
||||
* Sessions should be closed as soon as possible to prevent them holding memory
|
||||
and other resources when they have no more messages to handle
|
||||
* The closing API should allow the caller to detect when the pipeline is empty
|
||||
and it is safe to continue the WebSocket closing procedure
|
||||
* Individual extensions should remain as simple as possible to facilitate
|
||||
modularity and independent authorship
|
||||
|
||||
The final point about modularity is an important one: this framework is designed
|
||||
to facilitate extensions existing as plugins, by decoupling the protocol driver,
|
||||
extensions, and application. In an ideal world, plugins should only need to
|
||||
contain code for their specific functionality, and not solve these problems that
|
||||
apply to all sessions. Also, solving some of these problems requires
|
||||
consideration of all active sessions collectively, which an individual session
|
||||
is incapable of doing.
|
||||
|
||||
For example, it is entirely possible to take the simple `deflate` extension
|
||||
above and wrap its `incoming()` and `outgoing()` methods in two `Transform`
|
||||
streams, producing this type:
|
||||
|
||||
type Session = {
|
||||
incoming :: TransformStream
|
||||
outtoing :: TransformStream
|
||||
close :: () -> ()
|
||||
}
|
||||
|
||||
The `Transform` class makes it easy to wrap an async function such that message
|
||||
order is preserved:
|
||||
|
||||
```js
|
||||
var stream = require('stream'),
|
||||
session = new stream.Transform({ objectMode: true });
|
||||
|
||||
session._transform = function(message, _, callback) {
|
||||
var self = this;
|
||||
deflate.outgoing(message, function(error, result) {
|
||||
self.push(result);
|
||||
callback();
|
||||
});
|
||||
};
|
||||
```
|
||||
|
||||
However, this has a negative impact on throughput: it works by deferring
|
||||
`callback()` until the async function has 'returned', which blocks `Transform`
|
||||
from passing further input into the `_transform()` method until the current
|
||||
message is dealt with completely. This would prevent sessions from processing
|
||||
messages concurrently, and would unnecessarily reduce the throughput of
|
||||
stateless extensions.
|
||||
|
||||
So, input should be handed off to sessions as soon as possible, and all we need
|
||||
is a mechanism to reorder the output so that message order is preserved for the
|
||||
next session in line.
|
||||
|
||||
|
||||
## Solution
|
||||
|
||||
We now describe the model implemented here and how it meets the above design
|
||||
goals. The above diagram where a stack of extensions sit between the driver and
|
||||
application describes the data flow, but not the object graph. That looks like
|
||||
this:
|
||||
|
||||
|
||||
+--------+
|
||||
| Driver |
|
||||
+---o----+
|
||||
|
|
||||
V
|
||||
+------------+ +----------+
|
||||
| Extensions o----->| Pipeline |
|
||||
+------------+ +-----o----+
|
||||
|
|
||||
+---------------+---------------+
|
||||
| | |
|
||||
+-----o----+ +-----o----+ +-----o----+
|
||||
| Cell [A] | | Cell [B] | | Cell [C] |
|
||||
+----------+ +----------+ +----------+
|
||||
|
||||
|
||||
A driver using this framework holds an instance of the `Extensions` class, which
|
||||
it uses to register extension plugins, negotiate headers and transform messages.
|
||||
The `Extensions` instance itself holds a `Pipeline`, which contains an array of
|
||||
`Cell` objects, each of which wraps one of the sessions.
|
||||
|
||||
|
||||
### Message processing
|
||||
|
||||
Both the `Pipeline` and `Cell` classes have `incoming()` and `outgoing()`
|
||||
methods; the `Pipeline` interface pushes messages into the pipe, delegates the
|
||||
message to each `Cell` in turn, then returns it back to the driver. Outgoing
|
||||
messages pass through `A` then `B` then `C`, and incoming messages in the
|
||||
reverse order.
|
||||
|
||||
Internally, a `Cell` contains two `Functor` objects. A `Functor` wraps an async
|
||||
function and makes sure its output messages maintain the order of its input
|
||||
messages. This name is due to [@fronx](https://github.com/fronx), on the basis
|
||||
that, by preserving message order, the abstraction preserves the *mapping*
|
||||
between input and output messages. To use our simple `deflate` extension from
|
||||
above:
|
||||
|
||||
```js
|
||||
var functor = new Functor(deflate, 'outgoing');
|
||||
|
||||
functor.call({ data: large }, function() {
|
||||
console.log(1, 'large');
|
||||
});
|
||||
|
||||
functor.call({ data: small }, function() {
|
||||
console.log(2, 'small');
|
||||
});
|
||||
|
||||
/* -> 1 'large'
|
||||
2 'small' */
|
||||
```
|
||||
|
||||
A `Cell` contains two of these, one for each direction:
|
||||
|
||||
|
||||
+-----------------------+
|
||||
+---->| Functor [A, incoming] |
|
||||
+----------+ | +-----------------------+
|
||||
| Cell [A] o------+
|
||||
+----------+ | +-----------------------+
|
||||
+---->| Functor [A, outgoing] |
|
||||
+-----------------------+
|
||||
|
||||
|
||||
This satisfies the message transformation requirements: the `Pipeline` simply
|
||||
loops over the cells in the appropriate direction to transform each message.
|
||||
Because each `Cell` will preserve message order, we can pass a message to the
|
||||
next `Cell` in line as soon as the current `Cell` returns it. This gives each
|
||||
`Cell` all the messages in order while maximising throughput.
|
||||
|
||||
|
||||
### Session closing
|
||||
|
||||
We want to close each session as soon as possible, after all existing messages
|
||||
have drained. To do this, each `Cell` begins with a pending message counter in
|
||||
each direction, labelled `in` and `out` below.
|
||||
|
||||
|
||||
+----------+
|
||||
| Pipeline |
|
||||
+-----o----+
|
||||
|
|
||||
+---------------+---------------+
|
||||
| | |
|
||||
+-----o----+ +-----o----+ +-----o----+
|
||||
| Cell [A] | | Cell [B] | | Cell [C] |
|
||||
+----------+ +----------+ +----------+
|
||||
in: 0 in: 0 in: 0
|
||||
out: 0 out: 0 out: 0
|
||||
|
||||
|
||||
When a message *m1* enters the pipeline, say in the `outgoing` direction, we
|
||||
increment the `pending.out` counter on all cells immediately.
|
||||
|
||||
|
||||
+----------+
|
||||
m1 => | Pipeline |
|
||||
+-----o----+
|
||||
|
|
||||
+---------------+---------------+
|
||||
| | |
|
||||
+-----o----+ +-----o----+ +-----o----+
|
||||
| Cell [A] | | Cell [B] | | Cell [C] |
|
||||
+----------+ +----------+ +----------+
|
||||
in: 0 in: 0 in: 0
|
||||
out: 1 out: 1 out: 1
|
||||
|
||||
|
||||
*m1* is handed off to `A`, meanwhile a second message `m2` arrives in the same
|
||||
direction. All `pending.out` counters are again incremented.
|
||||
|
||||
|
||||
+----------+
|
||||
m2 => | Pipeline |
|
||||
+-----o----+
|
||||
|
|
||||
+---------------+---------------+
|
||||
m1 | | |
|
||||
+-----o----+ +-----o----+ +-----o----+
|
||||
| Cell [A] | | Cell [B] | | Cell [C] |
|
||||
+----------+ +----------+ +----------+
|
||||
in: 0 in: 0 in: 0
|
||||
out: 2 out: 2 out: 2
|
||||
|
||||
|
||||
When the first cell's `A.outgoing` functor finishes processing *m1*, the first
|
||||
`pending.out` counter is decremented and *m1* is handed off to cell `B`.
|
||||
|
||||
|
||||
+----------+
|
||||
| Pipeline |
|
||||
+-----o----+
|
||||
|
|
||||
+---------------+---------------+
|
||||
m2 | m1 | |
|
||||
+-----o----+ +-----o----+ +-----o----+
|
||||
| Cell [A] | | Cell [B] | | Cell [C] |
|
||||
+----------+ +----------+ +----------+
|
||||
in: 0 in: 0 in: 0
|
||||
out: 1 out: 2 out: 2
|
||||
|
||||
|
||||
|
||||
As `B` finishes with *m1*, and as `A` finishes with *m2*, the `pending.out`
|
||||
counters continue to decrement.
|
||||
|
||||
|
||||
+----------+
|
||||
| Pipeline |
|
||||
+-----o----+
|
||||
|
|
||||
+---------------+---------------+
|
||||
| m2 | m1 |
|
||||
+-----o----+ +-----o----+ +-----o----+
|
||||
| Cell [A] | | Cell [B] | | Cell [C] |
|
||||
+----------+ +----------+ +----------+
|
||||
in: 0 in: 0 in: 0
|
||||
out: 0 out: 1 out: 2
|
||||
|
||||
|
||||
|
||||
Say `C` is a little slow, and begins processing *m2* while still processing
|
||||
*m1*. That's fine, the `Functor` mechanism will keep *m1* ahead of *m2* in the
|
||||
output.
|
||||
|
||||
|
||||
+----------+
|
||||
| Pipeline |
|
||||
+-----o----+
|
||||
|
|
||||
+---------------+---------------+
|
||||
| | m2 | m1
|
||||
+-----o----+ +-----o----+ +-----o----+
|
||||
| Cell [A] | | Cell [B] | | Cell [C] |
|
||||
+----------+ +----------+ +----------+
|
||||
in: 0 in: 0 in: 0
|
||||
out: 0 out: 0 out: 2
|
||||
|
||||
|
||||
Once all messages are dealt with, the counters return to `0`.
|
||||
|
||||
|
||||
+----------+
|
||||
| Pipeline |
|
||||
+-----o----+
|
||||
|
|
||||
+---------------+---------------+
|
||||
| | |
|
||||
+-----o----+ +-----o----+ +-----o----+
|
||||
| Cell [A] | | Cell [B] | | Cell [C] |
|
||||
+----------+ +----------+ +----------+
|
||||
in: 0 in: 0 in: 0
|
||||
out: 0 out: 0 out: 0
|
||||
|
||||
|
||||
The same process applies in the `incoming` direction, the only difference being
|
||||
that messages are passed to `C` first.
|
||||
|
||||
This makes closing the sessions quite simple. When the driver wants to close the
|
||||
socket, it calls `Pipeline.close()`. This *immediately* calls `close()` on all
|
||||
the cells. If a cell has `in == out == 0`, then it immediately calls
|
||||
`session.close()`. Otherwise, it stores the closing call and defers it until
|
||||
`in` and `out` have both ticked down to zero. The pipeline will not accept new
|
||||
messages after `close()` has been called, so we know the pending counts will not
|
||||
increase after this point.
|
||||
|
||||
This means each session is closed as soon as possible: `A` can close while the
|
||||
slow `C` session is still working, because it knows there are no more messages
|
||||
on the way. Similarly, `C` will defer closing if `close()` is called while *m1*
|
||||
is still in `B`, and *m2* in `A`, because its pending count means it knows it
|
||||
has work yet to do, even if it's not received those messages yet. This concern
|
||||
cannot be addressed by extensions acting only on their own local state, unless
|
||||
we pollute individual extensions by making them all implement this same
|
||||
mechanism.
|
||||
|
||||
The actual closing API at each level is slightly different:
|
||||
|
||||
type Session = {
|
||||
close :: () -> ()
|
||||
}
|
||||
|
||||
type Cell = {
|
||||
close :: () -> Promise ()
|
||||
}
|
||||
|
||||
type Pipeline = {
|
||||
close :: Callback -> ()
|
||||
}
|
||||
|
||||
This might appear inconsistent so it's worth explaining. Remember that a
|
||||
`Pipeline` holds a list of `Cell` objects, each wrapping a `Session`. The driver
|
||||
talks (via the `Extensions` API) to the `Pipeline` interface, and it wants
|
||||
`Pipeline.close()` to do two things: close all the sessions, and tell me when
|
||||
it's safe to start the closing procedure (i.e. when all messages have drained
|
||||
from the pipe and been handed off to the application or socket). A callback API
|
||||
works well for that.
|
||||
|
||||
At the other end of the stack, `Session.close()` is a nullary void method with
|
||||
no callback or promise API because we don't care what it does, and whatever it
|
||||
does do will not block the WebSocket protocol; we're not going to hold off
|
||||
processing messages while a session closes its de/compression context. We just
|
||||
tell it to close itself, and don't want to wait while it does that.
|
||||
|
||||
In the middle, `Cell.close()` returns a promise rather than using a callback.
|
||||
This is for two reasons. First, `Cell.close()` might not do anything
|
||||
immediately, it might have to defer its effect while messages drain. So, if
|
||||
given a callback, it would have to store it in a queue for later execution.
|
||||
Callbacks work fine if your method does something and can then invoke the
|
||||
callback itself, but if you need to store callbacks somewhere so another method
|
||||
can execute them, a promise is a better fit. Second, it better serves the
|
||||
purposes of `Pipeline.close()`: it wants to call `close()` on each of a list of
|
||||
cells, and wait for all of them to finish. This is simple and idiomatic using
|
||||
promises:
|
||||
|
||||
```js
|
||||
var closed = cells.map((cell) => cell.close());
|
||||
Promise.all(closed).then(callback);
|
||||
```
|
||||
|
||||
(We don't actually use a full *Promises/A+* compatible promise here, we use a
|
||||
much simplified construction that acts as a callback aggregater and resolves
|
||||
synchronously and does not support chaining, but the principle is the same.)
|
||||
|
||||
|
||||
### Error handling
|
||||
|
||||
We've not mentioned error handling so far but it bears some explanation. The
|
||||
above counter system still applies, but behaves slightly differently in the
|
||||
presence of errors.
|
||||
|
||||
Say we push three messages into the pipe in the outgoing direction:
|
||||
|
||||
|
||||
+----------+
|
||||
m3, m2, m1 => | Pipeline |
|
||||
+-----o----+
|
||||
|
|
||||
+---------------+---------------+
|
||||
| | |
|
||||
+-----o----+ +-----o----+ +-----o----+
|
||||
| Cell [A] | | Cell [B] | | Cell [C] |
|
||||
+----------+ +----------+ +----------+
|
||||
in: 0 in: 0 in: 0
|
||||
out: 3 out: 3 out: 3
|
||||
|
||||
|
||||
They pass through the cells successfully up to this point:
|
||||
|
||||
|
||||
+----------+
|
||||
| Pipeline |
|
||||
+-----o----+
|
||||
|
|
||||
+---------------+---------------+
|
||||
m3 | m2 | m1 |
|
||||
+-----o----+ +-----o----+ +-----o----+
|
||||
| Cell [A] | | Cell [B] | | Cell [C] |
|
||||
+----------+ +----------+ +----------+
|
||||
in: 0 in: 0 in: 0
|
||||
out: 1 out: 2 out: 3
|
||||
|
||||
|
||||
At this point, session `B` produces an error while processing *m2*, that is *m2*
|
||||
becomes *e2*. *m1* is still in the pipeline, and *m3* is queued behind *m2*.
|
||||
What ought to happen is that *m1* is handed off to the socket, then *m2* is
|
||||
released to the driver, which will detect the error and begin closing the
|
||||
socket. No further processing should be done on *m3* and it should not be
|
||||
released to the driver after the error is emitted.
|
||||
|
||||
To handle this, we allow errors to pass down the pipeline just like messages do,
|
||||
to maintain ordering. But, once a cell sees its session produce an error, or it
|
||||
receives an error from upstream, it should refuse to accept any further
|
||||
messages. Session `B` might have begun processing *m3* by the time it produces
|
||||
the error *e2*, but `C` will have been given *e2* before it receives *m3*, and
|
||||
can simply drop *m3*.
|
||||
|
||||
Now, say *e2* reaches the slow session `C` while *m1* is still present,
|
||||
meanwhile *m3* has been dropped. `C` will never receive *m3* since it will have
|
||||
been dropped upstream. Under the present model, its `out` counter will be `3`
|
||||
but it is only going to emit two more values: *m1* and *e2*. In order for
|
||||
closing to work, we need to decrement `out` to reflect this. The situation
|
||||
should look like this:
|
||||
|
||||
|
||||
+----------+
|
||||
| Pipeline |
|
||||
+-----o----+
|
||||
|
|
||||
+---------------+---------------+
|
||||
| | e2 | m1
|
||||
+-----o----+ +-----o----+ +-----o----+
|
||||
| Cell [A] | | Cell [B] | | Cell [C] |
|
||||
+----------+ +----------+ +----------+
|
||||
in: 0 in: 0 in: 0
|
||||
out: 0 out: 0 out: 2
|
||||
|
||||
|
||||
When a cell sees its session emit an error, or when it receives an error from
|
||||
upstream, it sets its pending count in the appropriate direction to equal the
|
||||
number of messages it is *currently* processing. It will not accept any messages
|
||||
after it sees the error, so this will allow the counter to reach zero.
|
||||
|
||||
Note that while *e2* is in the pipeline, `Pipeline` should drop any further
|
||||
messages in the outgoing direction, but should continue to accept incoming
|
||||
messages. Until *e2* makes it out of the pipe to the driver, behind previous
|
||||
successful messages, the driver does not know an error has happened, and a
|
||||
message may arrive over the socket and make it all the way through the incoming
|
||||
pipe in the meantime. We only halt processing in the affected direction to avoid
|
||||
doing unnecessary work since messages arriving after an error should not be
|
||||
processed.
|
||||
|
||||
Some unnecessary work may happen, for example any messages already in the
|
||||
pipeline following *m2* will be processed by `A`, since it's upstream of the
|
||||
error. Those messages will be dropped by `B`.
|
||||
|
||||
|
||||
## Alternative ideas
|
||||
|
||||
I am considering implementing `Functor` as an object-mode transform stream
|
||||
rather than what is essentially an async function. Being object-mode, a stream
|
||||
would preserve message boundaries and would also possibly help address
|
||||
back-pressure. I'm not sure whether this would require external API changes so
|
||||
that such streams could be connected to the downstream driver's streams.
|
||||
|
||||
|
||||
## Acknowledgements
|
||||
|
||||
Credit is due to [@mnowster](https://github.com/mnowster) for helping with the
|
||||
design and to [@fronx](https://github.com/fronx) for helping name things.
|
||||
@@ -0,0 +1,53 @@
|
||||
'use strict';
|
||||
|
||||
var Functor = require('./functor'),
|
||||
Pledge = require('./pledge');
|
||||
|
||||
var Cell = function(tuple) {
|
||||
this._ext = tuple[0];
|
||||
this._session = tuple[1];
|
||||
|
||||
this._functors = {
|
||||
incoming: new Functor(this._session, 'processIncomingMessage'),
|
||||
outgoing: new Functor(this._session, 'processOutgoingMessage')
|
||||
};
|
||||
};
|
||||
|
||||
Cell.prototype.pending = function(direction) {
|
||||
var functor = this._functors[direction];
|
||||
if (!functor._stopped) functor.pending += 1;
|
||||
};
|
||||
|
||||
Cell.prototype.incoming = function(error, message, callback, context) {
|
||||
this._exec('incoming', error, message, callback, context);
|
||||
};
|
||||
|
||||
Cell.prototype.outgoing = function(error, message, callback, context) {
|
||||
this._exec('outgoing', error, message, callback, context);
|
||||
};
|
||||
|
||||
Cell.prototype.close = function() {
|
||||
this._closed = this._closed || new Pledge();
|
||||
this._doClose();
|
||||
return this._closed;
|
||||
};
|
||||
|
||||
Cell.prototype._exec = function(direction, error, message, callback, context) {
|
||||
this._functors[direction].call(error, message, function(err, msg) {
|
||||
if (err) err.message = this._ext.name + ': ' + err.message;
|
||||
callback.call(context, err, msg);
|
||||
this._doClose();
|
||||
}, this);
|
||||
};
|
||||
|
||||
Cell.prototype._doClose = function() {
|
||||
var fin = this._functors.incoming,
|
||||
fout = this._functors.outgoing;
|
||||
|
||||
if (!this._closed || fin.pending + fout.pending !== 0) return;
|
||||
if (this._session) this._session.close();
|
||||
this._session = null;
|
||||
this._closed.done();
|
||||
};
|
||||
|
||||
module.exports = Cell;
|
||||
@@ -0,0 +1,72 @@
|
||||
'use strict';
|
||||
|
||||
var RingBuffer = require('./ring_buffer');
|
||||
|
||||
var Functor = function(session, method) {
|
||||
this._session = session;
|
||||
this._method = method;
|
||||
this._queue = new RingBuffer(Functor.QUEUE_SIZE);
|
||||
this._stopped = false;
|
||||
this.pending = 0;
|
||||
};
|
||||
|
||||
Functor.QUEUE_SIZE = 8;
|
||||
|
||||
Functor.prototype.call = function(error, message, callback, context) {
|
||||
if (this._stopped) return;
|
||||
|
||||
var record = { error: error, message: message, callback: callback, context: context, done: false },
|
||||
called = false,
|
||||
self = this;
|
||||
|
||||
this._queue.push(record);
|
||||
|
||||
if (record.error) {
|
||||
record.done = true;
|
||||
this._stop();
|
||||
return this._flushQueue();
|
||||
}
|
||||
|
||||
var handler = function(err, msg) {
|
||||
if (!(called ^ (called = true))) return;
|
||||
|
||||
if (err) {
|
||||
self._stop();
|
||||
record.error = err;
|
||||
record.message = null;
|
||||
} else {
|
||||
record.message = msg;
|
||||
}
|
||||
|
||||
record.done = true;
|
||||
self._flushQueue();
|
||||
};
|
||||
|
||||
try {
|
||||
this._session[this._method](message, handler);
|
||||
} catch (err) {
|
||||
handler(err);
|
||||
}
|
||||
};
|
||||
|
||||
Functor.prototype._stop = function() {
|
||||
this.pending = this._queue.length;
|
||||
this._stopped = true;
|
||||
};
|
||||
|
||||
Functor.prototype._flushQueue = function() {
|
||||
var queue = this._queue, record;
|
||||
|
||||
while (queue.length > 0 && queue.peek().done) {
|
||||
record = queue.shift();
|
||||
if (record.error) {
|
||||
this.pending = 0;
|
||||
queue.clear();
|
||||
} else {
|
||||
this.pending -= 1;
|
||||
}
|
||||
record.callback.call(record.context, record.error, record.message);
|
||||
}
|
||||
};
|
||||
|
||||
module.exports = Functor;
|
||||
@@ -0,0 +1,47 @@
|
||||
'use strict';
|
||||
|
||||
var Cell = require('./cell'),
|
||||
Pledge = require('./pledge');
|
||||
|
||||
var Pipeline = function(sessions) {
|
||||
this._cells = sessions.map(function(session) { return new Cell(session) });
|
||||
this._stopped = { incoming: false, outgoing: false };
|
||||
};
|
||||
|
||||
Pipeline.prototype.processIncomingMessage = function(message, callback, context) {
|
||||
if (this._stopped.incoming) return;
|
||||
this._loop('incoming', this._cells.length - 1, -1, -1, message, callback, context);
|
||||
};
|
||||
|
||||
Pipeline.prototype.processOutgoingMessage = function(message, callback, context) {
|
||||
if (this._stopped.outgoing) return;
|
||||
this._loop('outgoing', 0, this._cells.length, 1, message, callback, context);
|
||||
};
|
||||
|
||||
Pipeline.prototype.close = function(callback, context) {
|
||||
this._stopped = { incoming: true, outgoing: true };
|
||||
|
||||
var closed = this._cells.map(function(a) { return a.close() });
|
||||
if (callback)
|
||||
Pledge.all(closed).then(function() { callback.call(context) });
|
||||
};
|
||||
|
||||
Pipeline.prototype._loop = function(direction, start, end, step, message, callback, context) {
|
||||
var cells = this._cells,
|
||||
n = cells.length,
|
||||
self = this;
|
||||
|
||||
while (n--) cells[n].pending(direction);
|
||||
|
||||
var pipe = function(index, error, msg) {
|
||||
if (index === end) return callback.call(context, error, msg);
|
||||
|
||||
cells[index][direction](error, msg, function(err, m) {
|
||||
if (err) self._stopped[direction] = true;
|
||||
pipe(index + step, err, m);
|
||||
});
|
||||
};
|
||||
pipe(start, null, message);
|
||||
};
|
||||
|
||||
module.exports = Pipeline;
|
||||
@@ -0,0 +1,37 @@
|
||||
'use strict';
|
||||
|
||||
var RingBuffer = require('./ring_buffer');
|
||||
|
||||
var Pledge = function() {
|
||||
this._complete = false;
|
||||
this._callbacks = new RingBuffer(Pledge.QUEUE_SIZE);
|
||||
};
|
||||
|
||||
Pledge.QUEUE_SIZE = 4;
|
||||
|
||||
Pledge.all = function(list) {
|
||||
var pledge = new Pledge(),
|
||||
pending = list.length,
|
||||
n = pending;
|
||||
|
||||
if (pending === 0) pledge.done();
|
||||
|
||||
while (n--) list[n].then(function() {
|
||||
pending -= 1;
|
||||
if (pending === 0) pledge.done();
|
||||
});
|
||||
return pledge;
|
||||
};
|
||||
|
||||
Pledge.prototype.then = function(callback) {
|
||||
if (this._complete) callback();
|
||||
else this._callbacks.push(callback);
|
||||
};
|
||||
|
||||
Pledge.prototype.done = function() {
|
||||
this._complete = true;
|
||||
var callbacks = this._callbacks, callback;
|
||||
while (callback = callbacks.shift()) callback();
|
||||
};
|
||||
|
||||
module.exports = Pledge;
|
||||
@@ -0,0 +1,66 @@
|
||||
'use strict';
|
||||
|
||||
var RingBuffer = function(bufferSize) {
|
||||
this._bufferSize = bufferSize;
|
||||
this.clear();
|
||||
};
|
||||
|
||||
RingBuffer.prototype.clear = function() {
|
||||
this._buffer = new Array(this._bufferSize);
|
||||
this._ringOffset = 0;
|
||||
this._ringSize = this._bufferSize;
|
||||
this._head = 0;
|
||||
this._tail = 0;
|
||||
this.length = 0;
|
||||
};
|
||||
|
||||
RingBuffer.prototype.push = function(value) {
|
||||
var expandBuffer = false,
|
||||
expandRing = false;
|
||||
|
||||
if (this._ringSize < this._bufferSize) {
|
||||
expandBuffer = (this._tail === 0);
|
||||
} else if (this._ringOffset === this._ringSize) {
|
||||
expandBuffer = true;
|
||||
expandRing = (this._tail === 0);
|
||||
}
|
||||
|
||||
if (expandBuffer) {
|
||||
this._tail = this._bufferSize;
|
||||
this._buffer = this._buffer.concat(new Array(this._bufferSize));
|
||||
this._bufferSize = this._buffer.length;
|
||||
|
||||
if (expandRing)
|
||||
this._ringSize = this._bufferSize;
|
||||
}
|
||||
|
||||
this._buffer[this._tail] = value;
|
||||
this.length += 1;
|
||||
if (this._tail < this._ringSize) this._ringOffset += 1;
|
||||
this._tail = (this._tail + 1) % this._bufferSize;
|
||||
};
|
||||
|
||||
RingBuffer.prototype.peek = function() {
|
||||
if (this.length === 0) return void 0;
|
||||
return this._buffer[this._head];
|
||||
};
|
||||
|
||||
RingBuffer.prototype.shift = function() {
|
||||
if (this.length === 0) return void 0;
|
||||
|
||||
var value = this._buffer[this._head];
|
||||
this._buffer[this._head] = void 0;
|
||||
this.length -= 1;
|
||||
this._ringOffset -= 1;
|
||||
|
||||
if (this._ringOffset === 0 && this.length > 0) {
|
||||
this._head = this._ringSize;
|
||||
this._ringOffset = this.length;
|
||||
this._ringSize = this._bufferSize;
|
||||
} else {
|
||||
this._head = (this._head + 1) % this._ringSize;
|
||||
}
|
||||
return value;
|
||||
};
|
||||
|
||||
module.exports = RingBuffer;
|
||||
@@ -9,7 +9,7 @@ var Extensions = function() {
|
||||
this._byName = {};
|
||||
this._inOrder = [];
|
||||
this._sessions = [];
|
||||
this._index = {}
|
||||
this._index = {};
|
||||
};
|
||||
|
||||
Extensions.MESSAGE_OPCODES = [1, 2];
|
||||
@@ -89,9 +89,9 @@ var instance = {
|
||||
},
|
||||
|
||||
generateResponse: function(header) {
|
||||
var offers = Parser.parseHeader(header),
|
||||
sessions = [],
|
||||
response = [];
|
||||
var sessions = [],
|
||||
response = [],
|
||||
offers = Parser.parseHeader(header);
|
||||
|
||||
this._inOrder.forEach(function(ext) {
|
||||
var offer = offers.byName(ext.name);
|
||||
@@ -112,7 +112,7 @@ var instance = {
|
||||
},
|
||||
|
||||
validFrameRsv: function(frame) {
|
||||
var allowed = {rsv1: false, rsv2: false, rsv3: false},
|
||||
var allowed = { rsv1: false, rsv2: false, rsv3: false },
|
||||
ext;
|
||||
|
||||
if (Extensions.MESSAGE_OPCODES.indexOf(frame.opcode) >= 0) {
|
||||
@@ -137,12 +137,9 @@ var instance = {
|
||||
this._pipeline.processOutgoingMessage(message, callback, context);
|
||||
},
|
||||
|
||||
close: function() {
|
||||
for (var i = 0, n = this._sessions.length; i < n; i++) {
|
||||
try {
|
||||
this._sessions[i][1].close();
|
||||
} catch (e) {}
|
||||
}
|
||||
close: function(callback, context) {
|
||||
if (!this._pipeline) return callback.call(context);
|
||||
this._pipeline.close(callback, context);
|
||||
},
|
||||
|
||||
_reserve: function(ext) {
|
||||
|
||||
+28
-19
@@ -1,20 +1,29 @@
|
||||
{ "name" : "websocket-extensions"
|
||||
, "description" : "Generic extension manager for WebSocket connections"
|
||||
, "homepage" : "http://github.com/faye/websocket-extensions-node"
|
||||
, "author" : "James Coglan <jcoglan@gmail.com> (http://jcoglan.com/)"
|
||||
, "keywords" : ["websocket"]
|
||||
, "license" : "MIT"
|
||||
|
||||
, "version" : "0.1.0"
|
||||
, "engines" : {"node": ">=0.6.0"}
|
||||
, "main" : "./lib/websocket_extensions"
|
||||
, "devDependencies" : {"jstest": ""}
|
||||
|
||||
, "scripts" : {"test": "jstest spec/runner.js"}
|
||||
|
||||
, "repository" : { "type" : "git"
|
||||
, "url" : "git://github.com/faye/websocket-extensions-node.git"
|
||||
}
|
||||
|
||||
, "bugs" : "http://github.com/faye/websocket-extensions-node/issues"
|
||||
{
|
||||
"name": "websocket-extensions",
|
||||
"description": "Generic extension manager for WebSocket connections",
|
||||
"homepage": "http://github.com/faye/websocket-extensions-node",
|
||||
"author": "James Coglan <jcoglan@gmail.com> (http://jcoglan.com/)",
|
||||
"keywords": [
|
||||
"websocket"
|
||||
],
|
||||
"license": "Apache-2.0",
|
||||
"version": "0.1.4",
|
||||
"engines": {
|
||||
"node": ">=0.8.0"
|
||||
},
|
||||
"files": [
|
||||
"lib"
|
||||
],
|
||||
"main": "./lib/websocket_extensions",
|
||||
"devDependencies": {
|
||||
"jstest": "*"
|
||||
},
|
||||
"scripts": {
|
||||
"test": "jstest spec/runner.js"
|
||||
},
|
||||
"repository": {
|
||||
"type": "git",
|
||||
"url": "git://github.com/faye/websocket-extensions-node.git"
|
||||
},
|
||||
"bugs": "http://github.com/faye/websocket-extensions-node/issues"
|
||||
}
|
||||
|
||||
+34
-18
@@ -20,53 +20,69 @@ test.describe("Parser", function() { with(this) {
|
||||
}})
|
||||
|
||||
it("parses one offer with no params", function() { with(this) {
|
||||
assertEqual( [{name: "a", params: {}}],
|
||||
assertEqual( [{ name: "a", params: {}}],
|
||||
parse('a') )
|
||||
}})
|
||||
|
||||
it("parses two offers with no params", function() { with(this) {
|
||||
assertEqual( [{name: "a", params: {}}, {name: "b", params: {}}],
|
||||
assertEqual( [{ name: "a", params: {}}, { name: "b", params: {}}],
|
||||
parse('a, b') )
|
||||
}})
|
||||
|
||||
it("parses a duplicate offer name", function() { with(this) {
|
||||
assertEqual( [{name: "a", params: {}}, {name: "a", params: {}}],
|
||||
assertEqual( [{ name: "a", params: {}}, { name: "a", params: {}}],
|
||||
parse('a, a') )
|
||||
}})
|
||||
|
||||
it("parses a flag", function() { with(this) {
|
||||
assertEqual( [{name: "a", params: {b: true}}],
|
||||
assertEqual( [{ name: "a", params: { b: true }}],
|
||||
parse('a; b') )
|
||||
}})
|
||||
|
||||
it("parses an unquoted param", function() { with(this) {
|
||||
assertEqual( [{name: "a", params: {b: 1}}],
|
||||
assertEqual( [{ name: "a", params: { b: 1 }}],
|
||||
parse('a; b=1') )
|
||||
}})
|
||||
|
||||
it("parses a quoted param", function() { with(this) {
|
||||
assertEqual( [{name: "a", params: {b: 'hi, "there'}}],
|
||||
assertEqual( [{ name: "a", params: { b: 'hi, "there' }}],
|
||||
parse('a; b="hi, \\"there"') )
|
||||
}})
|
||||
|
||||
it("parses multiple params", function() { with(this) {
|
||||
assertEqual( [{name: "a", params: {b: true, c: 1, d: 'hi'}}],
|
||||
assertEqual( [{ name: "a", params: { b: true, c: 1, d: 'hi' }}],
|
||||
parse('a; b; c=1; d="hi"') )
|
||||
}})
|
||||
|
||||
it("parses duplicate params", function() { with(this) {
|
||||
assertEqual( [{name: "a", params: {b: [true, 'hi'], c: 1}}],
|
||||
assertEqual( [{ name: "a", params: { b: [true, 'hi'], c: 1 }}],
|
||||
parse('a; b; c=1; b="hi"') )
|
||||
}})
|
||||
|
||||
it("parses multiple complex offers", function() { with(this) {
|
||||
assertEqual( [{name: "a", params: {b: 1}},
|
||||
{name: "c", params: {}},
|
||||
{name: "b", params: {d: true}},
|
||||
{name: "c", params: {e: ['hi, there', true]}},
|
||||
{name: "a", params: {b: true}}],
|
||||
assertEqual( [{ name: "a", params: { b: 1 }},
|
||||
{ name: "c", params: {}},
|
||||
{ name: "b", params: { d: true }},
|
||||
{ name: "c", params: { e: ['hi, there', true] }},
|
||||
{ name: "a", params: { b: true }}],
|
||||
parse('a; b=1, c, b; d, c; e="hi, there"; e, a; b') )
|
||||
}})
|
||||
|
||||
it("parses an extension name that shadows an Object property", function() { with(this) {
|
||||
assertEqual( [{ name: "hasOwnProperty", params: {}}],
|
||||
parse('hasOwnProperty') )
|
||||
}})
|
||||
|
||||
it("parses an extension param that shadows an Object property", function() { with(this) {
|
||||
var result = parse('foo; hasOwnProperty; x')[0]
|
||||
assertEqual( result.params.hasOwnProperty, true )
|
||||
}})
|
||||
|
||||
it("rejects a string missing its closing quote", function() { with(this) {
|
||||
assertThrows(SyntaxError, function() {
|
||||
parse('foo; bar="fooa\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a\\a')
|
||||
})
|
||||
}})
|
||||
}})
|
||||
|
||||
describe("serializeParams", function() { with(this) {
|
||||
@@ -75,23 +91,23 @@ test.describe("Parser", function() { with(this) {
|
||||
}})
|
||||
|
||||
it("serializes a flag", function() { with(this) {
|
||||
assertEqual( 'a; b', Parser.serializeParams('a', {b: true}) )
|
||||
assertEqual( 'a; b', Parser.serializeParams('a', { b: true }) )
|
||||
}})
|
||||
|
||||
it("serializes an unquoted param", function() { with(this) {
|
||||
assertEqual( 'a; b=42', Parser.serializeParams('a', {b: '42'}) )
|
||||
assertEqual( 'a; b=42', Parser.serializeParams('a', { b: '42' }) )
|
||||
}})
|
||||
|
||||
it("serializes a quoted param", function() { with(this) {
|
||||
assertEqual( 'a; b="hi, there"', Parser.serializeParams('a', {b: 'hi, there'}) )
|
||||
assertEqual( 'a; b="hi, there"', Parser.serializeParams('a', { b: 'hi, there' }) )
|
||||
}})
|
||||
|
||||
it("serializes multiple params", function() { with(this) {
|
||||
assertEqual( 'a; b; c=1; d=hi', Parser.serializeParams('a', {b: true, c: 1, d: 'hi'}) )
|
||||
assertEqual( 'a; b; c=1; d=hi', Parser.serializeParams('a', { b: true, c: 1, d: 'hi' }) )
|
||||
}})
|
||||
|
||||
it("serializes duplicate params", function() { with(this) {
|
||||
assertEqual( 'a; b; b=hi; c=1', Parser.serializeParams('a', {b: [true, 'hi'], c: 1}) )
|
||||
assertEqual( 'a; b; b=hi; c=1', Parser.serializeParams('a', { b: [true, 'hi'], c: 1 }) )
|
||||
}})
|
||||
}})
|
||||
}})
|
||||
|
||||
@@ -6,7 +6,7 @@ test.describe("Extensions", function() { with(this) {
|
||||
before(function() { with(this) {
|
||||
this.extensions = new Extensions()
|
||||
|
||||
this.ext = {name: "deflate", type: "permessage", rsv1: true, rsv2: false, rsv3: false}
|
||||
this.ext = { name: "deflate", type: "permessage", rsv1: true, rsv2: false, rsv3: false }
|
||||
this.session = {}
|
||||
}})
|
||||
|
||||
@@ -38,20 +38,20 @@ test.describe("Extensions", function() { with(this) {
|
||||
|
||||
describe("client sessions", function() { with(this) {
|
||||
before(function() { with(this) {
|
||||
this.offer = {mode: "compress"}
|
||||
this.offer = { mode: "compress" }
|
||||
stub(ext, "createClientSession").returns(session)
|
||||
stub(session, "generateOffer").returns(offer)
|
||||
extensions.add(ext)
|
||||
|
||||
this.conflict = {name: "tar", type: "permessage", rsv1: true, rsv2: false, rsv3: false}
|
||||
this.conflict = { name: "tar", type: "permessage", rsv1: true, rsv2: false, rsv3: false }
|
||||
this.conflictSession = {}
|
||||
stub(conflict, "createClientSession").returns(conflictSession)
|
||||
stub(conflictSession, "generateOffer").returns({gzip: true})
|
||||
stub(conflictSession, "generateOffer").returns({ gzip: true })
|
||||
|
||||
this.nonconflict = {name: "reverse", type: "permessage", rsv1: false, rsv2: true, rsv3: false}
|
||||
this.nonconflict = { name: "reverse", type: "permessage", rsv1: false, rsv2: true, rsv3: false }
|
||||
this.nonconflictSession = {}
|
||||
stub(nonconflict, "createClientSession").returns(nonconflictSession)
|
||||
stub(nonconflictSession, "generateOffer").returns({utf8: true})
|
||||
stub(nonconflictSession, "generateOffer").returns({ utf8: true })
|
||||
|
||||
stub(session, "activate").returns(true)
|
||||
stub(conflictSession, "activate").returns(true)
|
||||
@@ -133,18 +133,18 @@ test.describe("Extensions", function() { with(this) {
|
||||
}})
|
||||
|
||||
it("activates one session with a boolean param", function() { with(this) {
|
||||
expect(session, "activate").given({gzip: true}).exactly(1).returning(true)
|
||||
expect(session, "activate").given({ gzip: true }).exactly(1).returning(true)
|
||||
extensions.activate("deflate; gzip")
|
||||
}})
|
||||
|
||||
it("activates one session with a string param", function() { with(this) {
|
||||
expect(session, "activate").given({mode: "compress"}).exactly(1).returning(true)
|
||||
expect(session, "activate").given({ mode: "compress" }).exactly(1).returning(true)
|
||||
extensions.activate("deflate; mode=compress")
|
||||
}})
|
||||
|
||||
it("activates multiple sessions", function() { with(this) {
|
||||
expect(session, "activate").given({a: true}).exactly(1).returning(true)
|
||||
expect(nonconflictSession, "activate").given({b: true}).exactly(1).returning(true)
|
||||
expect(session, "activate").given({ a: true }).exactly(1).returning(true)
|
||||
expect(nonconflictSession, "activate").given({ b: true }).exactly(1).returning(true)
|
||||
extensions.activate("deflate; a, reverse; b")
|
||||
}})
|
||||
|
||||
@@ -180,17 +180,17 @@ test.describe("Extensions", function() { with(this) {
|
||||
it("processes messages in the reverse order given in the server's response", function() { with(this) {
|
||||
extensions.activate("deflate, reverse")
|
||||
|
||||
extensions.processIncomingMessage({frames: []}, function(error, message) {
|
||||
assertNull(error)
|
||||
extensions.processIncomingMessage({ frames: [] }, function(error, message) {
|
||||
assertNull( error )
|
||||
assertEqual( ["reverse", "deflate"], message.frames )
|
||||
})
|
||||
}})
|
||||
|
||||
it("yields an error if a session yields an error", function() { with(this) {
|
||||
extensions.activate("deflate")
|
||||
stub(session, "processIncomingMessage").yields([{message: "ENOENT"}])
|
||||
stub(session, "processIncomingMessage").yields([{ message: "ENOENT" }])
|
||||
|
||||
extensions.processIncomingMessage({frames: []}, function(error, message) {
|
||||
extensions.processIncomingMessage({ frames: [] }, function(error, message) {
|
||||
assertEqual( "deflate: ENOENT", error.message )
|
||||
assertNull( message )
|
||||
})
|
||||
@@ -198,11 +198,11 @@ test.describe("Extensions", function() { with(this) {
|
||||
|
||||
it("does not call sessions after one has yielded an error", function() { with(this) {
|
||||
extensions.activate("deflate, reverse")
|
||||
stub(nonconflictSession, "processIncomingMessage").yields([{message: "ENOENT"}])
|
||||
stub(nonconflictSession, "processIncomingMessage").yields([{ message: "ENOENT" }])
|
||||
|
||||
expect(session, "processIncomingMessage").exactly(0)
|
||||
|
||||
extensions.processIncomingMessage({frames: []}, function() {})
|
||||
extensions.processIncomingMessage({ frames: [] }, function() {})
|
||||
}})
|
||||
}})
|
||||
|
||||
@@ -223,6 +223,95 @@ test.describe("Extensions", function() { with(this) {
|
||||
})
|
||||
}})
|
||||
|
||||
describe("error handling", function() { with(this) {
|
||||
include(FakeClock)
|
||||
|
||||
sharedExamplesFor("handles errors", function() { with(this) {
|
||||
before(function() { with(this) {
|
||||
clock.stub()
|
||||
extensions.activate("deflate, reverse")
|
||||
|
||||
stub(session, "processOutgoingMessage", function(message, callback) {
|
||||
setTimeout(function() { callback(null, message.concat("a")) }, 100)
|
||||
})
|
||||
|
||||
stub(nonconflictSession, "processOutgoingMessage", function(message, callback) {
|
||||
setTimeout(function() { callback(null, message.concat("b")) }, 100)
|
||||
})
|
||||
|
||||
stub(nonconflictSession, "processIncomingMessage", function(message, callback) {
|
||||
if (message[0] === 5) return emitError(callback)
|
||||
setTimeout(function() { callback(null, message.concat("c")) }, 50)
|
||||
})
|
||||
|
||||
stub(session, "processIncomingMessage", function(message, callback) {
|
||||
setTimeout(function() { callback(null, message.concat("d")) }, 100)
|
||||
})
|
||||
|
||||
stub(session, "close")
|
||||
stub(nonconflictSession, "close")
|
||||
|
||||
this.messages = []
|
||||
|
||||
var push = function(error, message) {
|
||||
if (error) extensions.close(function() { messages.push("close") })
|
||||
messages.push(message)
|
||||
}
|
||||
|
||||
;[1, 2, 3].forEach(function(n) {
|
||||
extensions.processOutgoingMessage([n], push)
|
||||
})
|
||||
|
||||
;[4, 5, 6].forEach(function(n, i) {
|
||||
setTimeout(function() {
|
||||
extensions.processIncomingMessage([n], push)
|
||||
}, 20 * i)
|
||||
})
|
||||
|
||||
clock.tick(200)
|
||||
}})
|
||||
|
||||
it("allows the message before the error through to the end", function() { with(this) {
|
||||
assertEqual( [4, "c", "d"], messages[0] )
|
||||
}})
|
||||
|
||||
it("yields the error to the end of the pipeline", function() { with(this) {
|
||||
assertNull( messages[1] )
|
||||
}})
|
||||
|
||||
it("does not yield the message after the error", function() { with(this) {
|
||||
assertNotEqual( arrayIncluding([6, "c", "d"]), messages )
|
||||
}})
|
||||
|
||||
it("yields all the messages in the direction unaffected by the error", function() { with(this) {
|
||||
assertEqual( [1, "a", "b"], messages[2] )
|
||||
assertEqual( [2, "a", "b"], messages[3] )
|
||||
assertEqual( [3, "a", "b"], messages[4] )
|
||||
}})
|
||||
|
||||
it("closes after all messages are processed", function() { with(this) {
|
||||
assertEqual( "close", messages[5] )
|
||||
assertEqual( 6, messages.length )
|
||||
}})
|
||||
}})
|
||||
|
||||
describe("with a sync error", function() { with(this) {
|
||||
define("emitError", function(callback) {
|
||||
throw new Error("sync error")
|
||||
})
|
||||
|
||||
itShouldBehaveLike("handles errors")
|
||||
}})
|
||||
|
||||
describe("with an async error", function() { with(this) {
|
||||
define("emitError", function(callback) {
|
||||
setTimeout(function() { callback(new Error("async error"), null) }, 10)
|
||||
})
|
||||
|
||||
itShouldBehaveLike("handles errors")
|
||||
}})
|
||||
}})
|
||||
|
||||
describe("async processors", function() { with(this) {
|
||||
include(FakeClock)
|
||||
|
||||
@@ -247,19 +336,69 @@ test.describe("Extensions", function() { with(this) {
|
||||
extensions.activate("deflate, reverse")
|
||||
|
||||
var out = []
|
||||
extensions.processOutgoingMessage({frames: []}, function(error, message) { out.push(message) })
|
||||
extensions.processOutgoingMessage({frames: [1]}, function(error, message) { out.push(message) })
|
||||
extensions.processOutgoingMessage({ frames: [] }, function(error, message) { out.push(message) })
|
||||
extensions.processOutgoingMessage({ frames: [1] }, function(error, message) { out.push(message) })
|
||||
clock.tick(200)
|
||||
|
||||
assertEqual( [{frames: ["a", "c"]}, {frames: [1, "b", "d"]}], out )
|
||||
assertEqual( [{ frames: ["a", "c"] }, { frames: [1, "b", "d"] }], out )
|
||||
}})
|
||||
|
||||
it("defers closing until the extension has finished processing", function() { with(this) {
|
||||
extensions.activate("deflate")
|
||||
|
||||
var closed = false, notified = false
|
||||
stub(session, "close", function() { closed = true })
|
||||
|
||||
extensions.processOutgoingMessage({ frames: [] }, function() {})
|
||||
extensions.close(function() { notified = true })
|
||||
|
||||
clock.tick(50)
|
||||
assertNot( closed || notified )
|
||||
|
||||
clock.tick(50)
|
||||
assert( closed && notified )
|
||||
}})
|
||||
|
||||
it("closes each session as soon as it finishes processing", function() { with(this) {
|
||||
extensions.activate("deflate, reverse")
|
||||
|
||||
var closed = [false, false], notified = false
|
||||
stub(session, "close", function() { closed[0] = true })
|
||||
stub(nonconflictSession, "close", function() { closed[1] = true })
|
||||
|
||||
extensions.processOutgoingMessage({ frames: [] }, function() {});
|
||||
extensions.close(function() { notified = true })
|
||||
|
||||
clock.tick(50)
|
||||
assertNot( closed[0] || closed[1] || notified )
|
||||
|
||||
clock.tick(100)
|
||||
assert( closed[0] )
|
||||
assertNot( closed[1] || notified )
|
||||
|
||||
clock.tick(50)
|
||||
assert( closed[0] && closed[1] && notified )
|
||||
}})
|
||||
|
||||
it("notifies of closure immeidately if already closed", function() { with(this) {
|
||||
extensions.activate("deflate")
|
||||
stub(session, "close", function() { closed = true })
|
||||
|
||||
extensions.processOutgoingMessage({ frames: [] }, function() {})
|
||||
extensions.close()
|
||||
clock.tick(100)
|
||||
|
||||
var notified = false
|
||||
extensions.close(function() { notified = true })
|
||||
assert( notified )
|
||||
}})
|
||||
}})
|
||||
|
||||
it("processes messages in the order given in the server's response", function() { with(this) {
|
||||
extensions.activate("deflate, reverse")
|
||||
|
||||
extensions.processOutgoingMessage({frames: []}, function(error, message) {
|
||||
assertNull(error)
|
||||
extensions.processOutgoingMessage({ frames: [] }, function(error, message) {
|
||||
assertNull( error )
|
||||
assertEqual( ["deflate", "reverse"], message.frames )
|
||||
})
|
||||
}})
|
||||
@@ -267,17 +406,17 @@ test.describe("Extensions", function() { with(this) {
|
||||
it("processes messages in the server's order, not the client's order", function() { with(this) {
|
||||
extensions.activate("reverse, deflate")
|
||||
|
||||
extensions.processOutgoingMessage({frames: []}, function(error, message) {
|
||||
assertNull(error)
|
||||
extensions.processOutgoingMessage({ frames: [] }, function(error, message) {
|
||||
assertNull( error )
|
||||
assertEqual( ["reverse", "deflate"], message.frames )
|
||||
})
|
||||
}})
|
||||
|
||||
it("yields an error if a session yields an error", function() { with(this) {
|
||||
extensions.activate("deflate")
|
||||
stub(session, "processOutgoingMessage").yields([{message: "ENOENT"}])
|
||||
stub(session, "processOutgoingMessage").yields([{ message: "ENOENT" }])
|
||||
|
||||
extensions.processOutgoingMessage({frames: []}, function(error, message) {
|
||||
extensions.processOutgoingMessage({ frames: [] }, function(error, message) {
|
||||
assertEqual( "deflate: ENOENT", error.message )
|
||||
assertNull( message )
|
||||
})
|
||||
@@ -285,30 +424,30 @@ test.describe("Extensions", function() { with(this) {
|
||||
|
||||
it("does not call sessions after one has yielded an error", function() { with(this) {
|
||||
extensions.activate("deflate, reverse")
|
||||
stub(session, "processOutgoingMessage").yields([{message: "ENOENT"}])
|
||||
stub(session, "processOutgoingMessage").yields([{ message: "ENOENT" }])
|
||||
|
||||
expect(nonconflictSession, "processOutgoingMessage").exactly(0)
|
||||
|
||||
extensions.processOutgoingMessage({frames: []}, function() {})
|
||||
extensions.processOutgoingMessage({ frames: [] }, function() {})
|
||||
}})
|
||||
}})
|
||||
}})
|
||||
|
||||
describe("server sessions", function() { with(this) {
|
||||
before(function() { with(this) {
|
||||
this.response = {mode: "compress"}
|
||||
this.response = { mode: "compress" }
|
||||
stub(ext, "createServerSession").returns(session)
|
||||
stub(session, "generateResponse").returns(response)
|
||||
|
||||
this.conflict = {name: "tar", type: "permessage", rsv1: true, rsv2: false, rsv3: false}
|
||||
this.conflict = { name: "tar", type: "permessage", rsv1: true, rsv2: false, rsv3: false }
|
||||
this.conflictSession = {}
|
||||
stub(conflict, "createServerSession").returns(conflictSession)
|
||||
stub(conflictSession, "generateResponse").returns({gzip: true})
|
||||
stub(conflictSession, "generateResponse").returns({ gzip: true })
|
||||
|
||||
this.nonconflict = {name: "reverse", type: "permessage", rsv1: false, rsv2: true, rsv3: false}
|
||||
this.nonconflict = { name: "reverse", type: "permessage", rsv1: false, rsv2: true, rsv3: false }
|
||||
this.nonconflictSession = {}
|
||||
stub(nonconflict, "createServerSession").returns(nonconflictSession)
|
||||
stub(nonconflictSession, "generateResponse").returns({utf8: true})
|
||||
stub(nonconflictSession, "generateResponse").returns({ utf8: true })
|
||||
|
||||
extensions.add(ext)
|
||||
extensions.add(conflict)
|
||||
@@ -317,12 +456,12 @@ test.describe("Extensions", function() { with(this) {
|
||||
|
||||
describe("generateResponse", function() { with(this) {
|
||||
it("asks the extension for a server session with the offer", function() { with(this) {
|
||||
expect(ext, "createServerSession").given([{flag: true}]).exactly(1).returning(session)
|
||||
expect(ext, "createServerSession").given([{ flag: true }]).exactly(1).returning(session)
|
||||
extensions.generateResponse("deflate; flag")
|
||||
}})
|
||||
|
||||
it("asks the extension for a server session with multiple offers", function() { with(this) {
|
||||
expect(ext, "createServerSession").given([{a: true}, {b: true}]).exactly(1).returning(session)
|
||||
expect(ext, "createServerSession").given([{ a: true }, { b: true }]).exactly(1).returning(session)
|
||||
extensions.generateResponse("deflate; a, deflate; b")
|
||||
}})
|
||||
|
||||
@@ -373,7 +512,11 @@ test.describe("Extensions", function() { with(this) {
|
||||
assertEqual( "deflate; mode=compress", extensions.generateResponse("deflate, tar") )
|
||||
}})
|
||||
|
||||
it("returns a response for potentially conflicting extensions if their preceeding extensions don't build a session", function() { with(this) {
|
||||
it("throws an error if the header is invalid", function() { with(this) {
|
||||
assertThrows(SyntaxError, function() { extensions.generateResponse("x-webkit- -frame") })
|
||||
}})
|
||||
|
||||
it("returns a response for potentially conflicting extensions if their preceding extensions don't build a session", function() { with(this) {
|
||||
stub(ext, "createServerSession").returns(null)
|
||||
assertEqual( "tar; gzip", extensions.generateResponse("deflate, tar") )
|
||||
}})
|
||||
|
||||
Reference in New Issue
Block a user