'use strict'
|
|
var util = require('util')
|
var transport = require('../spdy-transport')
|
|
var debug = {
|
server: require('debug')('spdy:connection:server'),
|
client: require('debug')('spdy:connection:client')
|
}
|
var EventEmitter = require('events').EventEmitter
|
|
var Stream = transport.Stream
|
|
function Connection (socket, options) {
|
EventEmitter.call(this)
|
|
var state = {}
|
this._spdyState = state
|
|
// NOTE: There's a big trick here. Connection is used as a `this` argument
|
// to the wrapped `connection` event listener.
|
// socket end doesn't necessarly mean connection drop
|
this.httpAllowHalfOpen = true
|
|
state.timeout = new transport.utils.Timeout(this)
|
|
// Protocol info
|
state.protocol = transport.protocol[options.protocol]
|
state.version = null
|
state.constants = state.protocol.constants
|
state.pair = null
|
state.isServer = options.isServer
|
|
// Root of priority tree (i.e. stream id = 0)
|
state.priorityRoot = new transport.Priority({
|
defaultWeight: state.constants.DEFAULT_WEIGHT,
|
maxCount: transport.protocol.base.constants.MAX_PRIORITY_STREAMS
|
})
|
|
// Defaults
|
state.maxStreams = options.maxStreams ||
|
state.constants.MAX_CONCURRENT_STREAMS
|
|
state.autoSpdy31 = options.protocol.name !== 'h2' && options.autoSpdy31
|
state.acceptPush = options.acceptPush === undefined
|
? !state.isServer
|
: options.acceptPush
|
|
if (options.maxChunk === false) { state.maxChunk = Infinity } else if (options.maxChunk === undefined) { state.maxChunk = transport.protocol.base.constants.DEFAULT_MAX_CHUNK } else {
|
state.maxChunk = options.maxChunk
|
}
|
|
// Connection-level flow control
|
var windowSize = options.windowSize || 1 << 20
|
state.window = new transport.Window({
|
id: 0,
|
isServer: state.isServer,
|
recv: {
|
size: state.constants.DEFAULT_WINDOW,
|
max: state.constants.MAX_INITIAL_WINDOW_SIZE
|
},
|
send: {
|
size: state.constants.DEFAULT_WINDOW,
|
max: state.constants.MAX_INITIAL_WINDOW_SIZE
|
}
|
})
|
|
// It starts with DEFAULT_WINDOW, update must be sent to change it on client
|
state.window.recv.setMax(windowSize)
|
|
// Boilerplate for Stream constructor
|
state.streamWindow = new transport.Window({
|
id: -1,
|
isServer: state.isServer,
|
recv: {
|
size: windowSize,
|
max: state.constants.MAX_INITIAL_WINDOW_SIZE
|
},
|
send: {
|
size: state.constants.DEFAULT_WINDOW,
|
max: state.constants.MAX_INITIAL_WINDOW_SIZE
|
}
|
})
|
|
// Various state info
|
state.pool = state.protocol.compressionPool.create(options.headerCompression)
|
state.counters = {
|
push: 0,
|
stream: 0
|
}
|
|
// Init streams list
|
state.stream = {
|
map: {},
|
count: 0,
|
nextId: state.isServer ? 2 : 1,
|
lastId: {
|
both: 0,
|
received: 0
|
}
|
}
|
state.ping = {
|
nextId: state.isServer ? 2 : 1,
|
map: {}
|
}
|
state.goaway = false
|
|
// Debug
|
state.debug = state.isServer ? debug.server : debug.client
|
|
// X-Forwarded feature
|
state.xForward = null
|
|
// Create parser and hole for framer
|
state.parser = state.protocol.parser.create({
|
// NOTE: needed to distinguish ping from ping ACK in SPDY
|
isServer: state.isServer,
|
window: state.window
|
})
|
state.framer = state.protocol.framer.create({
|
window: state.window,
|
timeout: state.timeout
|
})
|
|
// SPDY has PUSH enabled on servers
|
if (state.protocol.name === 'spdy') {
|
state.framer.enablePush(state.isServer)
|
}
|
|
if (!state.isServer) { state.parser.skipPreface() }
|
|
this.socket = socket
|
|
this._init()
|
}
|
util.inherits(Connection, EventEmitter)
|
exports.Connection = Connection
|
|
Connection.create = function create (socket, options) {
|
return new Connection(socket, options)
|
}
|
|
Connection.prototype._init = function init () {
|
var self = this
|
var state = this._spdyState
|
var pool = state.pool
|
|
// Initialize session window
|
state.window.recv.on('drain', function () {
|
self._onSessionWindowDrain()
|
})
|
|
// Initialize parser
|
state.parser.on('data', function (frame) {
|
self._handleFrame(frame)
|
})
|
state.parser.once('version', function (version) {
|
self._onVersion(version)
|
})
|
|
// Propagate parser errors
|
state.parser.on('error', function (err) {
|
self._onParserError(err)
|
})
|
|
// Propagate framer errors
|
state.framer.on('error', function (err) {
|
self.emit('error', err)
|
})
|
|
this.socket.pipe(state.parser)
|
state.framer.pipe(this.socket)
|
|
// Allow high-level api to catch socket errors
|
this.socket.on('error', function onSocketError (e) {
|
self.emit('error', e)
|
})
|
|
this.socket.once('close', function onclose (hadError) {
|
var err
|
if (hadError) {
|
err = new Error('socket hang up')
|
err.code = 'ECONNRESET'
|
}
|
|
self.destroyStreams(err)
|
self.emit('close')
|
|
if (state.pair) {
|
pool.put(state.pair)
|
}
|
|
state.framer.resume()
|
})
|
|
// Reset timeout on close
|
this.once('close', function () {
|
self.setTimeout(0)
|
})
|
|
function _onWindowOverflow () {
|
self._onWindowOverflow()
|
}
|
|
state.window.recv.on('overflow', _onWindowOverflow)
|
state.window.send.on('overflow', _onWindowOverflow)
|
|
// Do not allow half-open connections
|
this.socket.allowHalfOpen = false
|
}
|
|
Connection.prototype._onVersion = function _onVersion (version) {
|
var state = this._spdyState
|
var prev = state.version
|
var parser = state.parser
|
var framer = state.framer
|
var pool = state.pool
|
|
state.version = version
|
state.debug('id=0 version=%d', version)
|
|
// Ignore transition to 3.1
|
if (!prev) {
|
state.pair = pool.get(version)
|
parser.setCompression(state.pair)
|
framer.setCompression(state.pair)
|
}
|
framer.setVersion(version)
|
|
if (!state.isServer) {
|
framer.prefaceFrame()
|
if (state.xForward !== null) {
|
framer.xForwardedFor({ host: state.xForward })
|
}
|
}
|
|
// Send preface+settings frame (once)
|
framer.settingsFrame({
|
max_header_list_size: state.constants.DEFAULT_MAX_HEADER_LIST_SIZE,
|
max_concurrent_streams: state.maxStreams,
|
enable_push: state.acceptPush ? 1 : 0,
|
initial_window_size: state.window.recv.max
|
})
|
|
// Update session window
|
if (state.version >= 3.1 || (state.isServer && state.autoSpdy31)) { this._onSessionWindowDrain() }
|
|
this.emit('version', version)
|
}
|
|
Connection.prototype._onParserError = function _onParserError (err) {
|
var state = this._spdyState
|
|
// Prevent further errors
|
state.parser.kill()
|
|
// Send GOAWAY
|
if (err instanceof transport.protocol.base.utils.ProtocolError) {
|
this._goaway({
|
lastId: state.stream.lastId.both,
|
code: err.code,
|
extra: err.message,
|
send: true
|
})
|
}
|
|
this.emit('error', err)
|
}
|
|
Connection.prototype._handleFrame = function _handleFrame (frame) {
|
var state = this._spdyState
|
|
state.debug('id=0 frame', frame)
|
state.timeout.reset()
|
|
// For testing purposes
|
this.emit('frame', frame)
|
|
var stream
|
|
// Session window update
|
if (frame.type === 'WINDOW_UPDATE' && frame.id === 0) {
|
if (state.version < 3.1 && state.autoSpdy31) {
|
state.debug('id=0 switch version to 3.1')
|
state.version = 3.1
|
this.emit('version', 3.1)
|
}
|
state.window.send.update(frame.delta)
|
return
|
}
|
|
if (state.isServer && frame.type === 'PUSH_PROMISE') {
|
state.debug('id=0 server PUSH_PROMISE')
|
this._goaway({
|
lastId: state.stream.lastId.both,
|
code: 'PROTOCOL_ERROR',
|
send: true
|
})
|
return
|
}
|
|
if (!stream && frame.id !== undefined) {
|
// Load created one
|
stream = state.stream.map[frame.id]
|
|
// Fail if not found
|
if (!stream &&
|
frame.type !== 'HEADERS' &&
|
frame.type !== 'PRIORITY' &&
|
frame.type !== 'RST') {
|
// Other side should destroy the stream upon receiving GOAWAY
|
if (this._isGoaway(frame.id)) { return }
|
|
state.debug('id=0 stream=%d not found', frame.id)
|
state.framer.rstFrame({ id: frame.id, code: 'INVALID_STREAM' })
|
return
|
}
|
}
|
|
// Create new stream
|
if (!stream && frame.type === 'HEADERS') {
|
this._handleHeaders(frame)
|
return
|
}
|
|
if (stream) {
|
stream._handleFrame(frame)
|
} else if (frame.type === 'SETTINGS') {
|
this._handleSettings(frame.settings)
|
} else if (frame.type === 'ACK_SETTINGS') {
|
// TODO(indutny): handle it one day
|
} else if (frame.type === 'PING') {
|
this._handlePing(frame)
|
} else if (frame.type === 'GOAWAY') {
|
this._handleGoaway(frame)
|
} else if (frame.type === 'X_FORWARDED_FOR') {
|
// Set X-Forwarded-For only once
|
if (state.xForward === null) {
|
state.xForward = frame.host
|
}
|
} else if (frame.type === 'PRIORITY') {
|
// TODO(indutny): handle this
|
} else {
|
state.debug('id=0 unknown frame type: %s', frame.type)
|
}
|
}
|
|
Connection.prototype._onWindowOverflow = function _onWindowOverflow () {
|
var state = this._spdyState
|
state.debug('id=0 window overflow')
|
this._goaway({
|
lastId: state.stream.lastId.both,
|
code: 'FLOW_CONTROL_ERROR',
|
send: true
|
})
|
}
|
|
Connection.prototype._isGoaway = function _isGoaway (id) {
|
var state = this._spdyState
|
if (state.goaway !== false && state.goaway < id) { return true }
|
return false
|
}
|
|
Connection.prototype._getId = function _getId () {
|
var state = this._spdyState
|
|
var id = state.stream.nextId
|
state.stream.nextId += 2
|
return id
|
}
|
|
Connection.prototype._createStream = function _createStream (uri) {
|
var state = this._spdyState
|
var id = uri.id
|
if (id === undefined) { id = this._getId() }
|
|
var isGoaway = this._isGoaway(id)
|
|
if (uri.push && !state.acceptPush) {
|
state.debug('id=0 push disabled promisedId=%d', id)
|
|
// Fatal error
|
this._goaway({
|
lastId: state.stream.lastId.both,
|
code: 'PROTOCOL_ERROR',
|
send: true
|
})
|
isGoaway = true
|
}
|
|
var stream = new Stream(this, {
|
id: id,
|
request: uri.request !== false,
|
method: uri.method,
|
path: uri.path,
|
host: uri.host,
|
priority: uri.priority,
|
headers: uri.headers,
|
parent: uri.parent,
|
readable: !isGoaway && uri.readable,
|
writable: !isGoaway && uri.writable
|
})
|
var self = this
|
|
// Just an empty stream for API consistency
|
if (isGoaway) {
|
return stream
|
}
|
|
state.stream.lastId.both = Math.max(state.stream.lastId.both, id)
|
|
state.debug('id=0 add stream=%d', stream.id)
|
state.stream.map[stream.id] = stream
|
state.stream.count++
|
state.counters.stream++
|
if (stream.parent !== null) {
|
state.counters.push++
|
}
|
|
stream.once('close', function () {
|
self._removeStream(stream)
|
})
|
|
return stream
|
}
|
|
Connection.prototype._handleHeaders = function _handleHeaders (frame) {
|
var state = this._spdyState
|
|
// Must be HEADERS frame after stream close
|
if (frame.id <= state.stream.lastId.received) { return }
|
|
// Someone is using our ids!
|
if ((frame.id + state.stream.nextId) % 2 === 0) {
|
state.framer.rstFrame({ id: frame.id, code: 'PROTOCOL_ERROR' })
|
return
|
}
|
|
var stream = this._createStream({
|
id: frame.id,
|
request: false,
|
method: frame.headers[':method'],
|
path: frame.headers[':path'],
|
host: frame.headers[':authority'],
|
priority: frame.priority,
|
headers: frame.headers,
|
writable: frame.writable
|
})
|
|
// GOAWAY
|
if (this._isGoaway(stream.id)) {
|
return
|
}
|
|
state.stream.lastId.received = Math.max(
|
state.stream.lastId.received,
|
stream.id
|
)
|
|
// TODO(indutny) handle stream limit
|
if (!this.emit('stream', stream)) {
|
// No listeners was set - abort the stream
|
stream.abort()
|
return
|
}
|
|
// Create fake frame to simulate end of the data
|
if (frame.fin) {
|
stream._handleFrame({ type: 'FIN', fin: true })
|
}
|
|
return stream
|
}
|
|
Connection.prototype._onSessionWindowDrain = function _onSessionWindowDrain () {
|
var state = this._spdyState
|
if (state.version < 3.1 && !(state.isServer && state.autoSpdy31)) {
|
return
|
}
|
|
var delta = state.window.recv.getDelta()
|
if (delta === 0) {
|
return
|
}
|
|
state.debug('id=0 session window drain, update by %d', delta)
|
|
state.framer.windowUpdateFrame({
|
id: 0,
|
delta: delta
|
})
|
state.window.recv.update(delta)
|
}
|
|
Connection.prototype.start = function start (version) {
|
this._spdyState.parser.setVersion(version)
|
}
|
|
// Mostly for testing
|
Connection.prototype.getVersion = function getVersion () {
|
return this._spdyState.version
|
}
|
|
Connection.prototype._handleSettings = function _handleSettings (settings) {
|
var state = this._spdyState
|
|
state.framer.ackSettingsFrame()
|
|
this._setDefaultWindow(settings)
|
if (settings.max_frame_size) { state.framer.setMaxFrameSize(settings.max_frame_size) }
|
|
// TODO(indutny): handle max_header_list_size
|
if (settings.header_table_size) {
|
try {
|
state.pair.compress.updateTableSize(settings.header_table_size)
|
} catch (e) {
|
this._goaway({
|
lastId: 0,
|
code: 'PROTOCOL_ERROR',
|
send: true
|
})
|
return
|
}
|
}
|
|
// HTTP2 clients needs to enable PUSH streams explicitly
|
if (state.protocol.name !== 'spdy') {
|
if (settings.enable_push === undefined) {
|
state.framer.enablePush(state.isServer)
|
} else {
|
state.framer.enablePush(settings.enable_push === 1)
|
}
|
}
|
|
// TODO(indutny): handle max_concurrent_streams
|
}
|
|
Connection.prototype._setDefaultWindow = function _setDefaultWindow (settings) {
|
if (settings.initial_window_size === undefined) {
|
return
|
}
|
|
var state = this._spdyState
|
|
// Update defaults
|
var window = state.streamWindow
|
window.send.setMax(settings.initial_window_size)
|
|
// Update existing streams
|
Object.keys(state.stream.map).forEach(function (id) {
|
var stream = state.stream.map[id]
|
var window = stream._spdyState.window
|
|
window.send.updateMax(settings.initial_window_size)
|
})
|
}
|
|
Connection.prototype._handlePing = function handlePing (frame) {
|
var self = this
|
var state = this._spdyState
|
|
// Handle incoming PING
|
if (!frame.ack) {
|
state.framer.pingFrame({
|
opaque: frame.opaque,
|
ack: true
|
})
|
|
self.emit('ping', frame.opaque)
|
return
|
}
|
|
// Handle reply PING
|
var hex = frame.opaque.toString('hex')
|
if (!state.ping.map[hex]) {
|
return
|
}
|
var ping = state.ping.map[hex]
|
delete state.ping.map[hex]
|
|
if (ping.cb) {
|
ping.cb(null)
|
}
|
}
|
|
Connection.prototype._handleGoaway = function handleGoaway (frame) {
|
this._goaway({
|
lastId: frame.lastId,
|
code: frame.code,
|
send: false
|
})
|
}
|
|
Connection.prototype.ping = function ping (callback) {
|
var state = this._spdyState
|
|
// HTTP2 is using 8-byte opaque
|
var opaque = Buffer.alloc(state.constants.PING_OPAQUE_SIZE)
|
opaque.fill(0)
|
opaque.writeUInt32BE(state.ping.nextId, opaque.length - 4)
|
state.ping.nextId += 2
|
|
state.ping.map[opaque.toString('hex')] = { cb: callback }
|
state.framer.pingFrame({
|
opaque: opaque,
|
ack: false
|
})
|
}
|
|
Connection.prototype.getCounter = function getCounter (name) {
|
return this._spdyState.counters[name]
|
}
|
|
Connection.prototype.reserveStream = function reserveStream (uri, callback) {
|
var stream = this._createStream(uri)
|
|
// GOAWAY
|
if (this._isGoaway(stream.id)) {
|
var err = new Error('Can\'t send request after GOAWAY')
|
process.nextTick(function () {
|
if (callback) { callback(err) } else {
|
stream.emit('error', err)
|
}
|
})
|
return stream
|
}
|
|
if (callback) {
|
process.nextTick(function () {
|
callback(null, stream)
|
})
|
}
|
|
return stream
|
}
|
|
Connection.prototype.request = function request (uri, callback) {
|
var stream = this.reserveStream(uri, function (err) {
|
if (err) {
|
if (callback) {
|
callback(err)
|
} else {
|
stream.emit('error', err)
|
}
|
return
|
}
|
|
if (stream._wasSent()) {
|
if (callback) {
|
callback(null, stream)
|
}
|
return
|
}
|
|
stream.send(function (err) {
|
if (err) {
|
if (callback) { return callback(err) } else { return stream.emit('error', err) }
|
}
|
|
if (callback) {
|
callback(null, stream)
|
}
|
})
|
})
|
|
return stream
|
}
|
|
Connection.prototype._removeStream = function _removeStream (stream) {
|
var state = this._spdyState
|
|
state.debug('id=0 remove stream=%d', stream.id)
|
delete state.stream.map[stream.id]
|
state.stream.count--
|
|
if (state.stream.count === 0) {
|
this.emit('_streamDrain')
|
}
|
}
|
|
Connection.prototype._goaway = function _goaway (params) {
|
var state = this._spdyState
|
var self = this
|
|
state.goaway = params.lastId
|
state.debug('id=0 goaway from=%d', state.goaway)
|
|
Object.keys(state.stream.map).forEach(function (id) {
|
var stream = state.stream.map[id]
|
|
// Abort every stream started after GOAWAY
|
if (stream.id <= params.lastId) {
|
return
|
}
|
|
stream.abort()
|
stream.emit('error', new Error('New stream after GOAWAY'))
|
})
|
|
function finish () {
|
// Destroy socket if there are no streams
|
if (state.stream.count === 0 || params.code !== 'OK') {
|
// No further frames should be processed
|
state.parser.kill()
|
|
process.nextTick(function () {
|
var err = new Error('Fatal error: ' + params.code)
|
self._onStreamDrain(err)
|
})
|
return
|
}
|
|
self.on('_streamDrain', self._onStreamDrain)
|
}
|
|
if (params.send) {
|
// Make sure that GOAWAY frame is sent before dumping framer
|
state.framer.goawayFrame({
|
lastId: params.lastId,
|
code: params.code,
|
extra: params.extra
|
}, finish)
|
} else {
|
finish()
|
}
|
}
|
|
Connection.prototype._onStreamDrain = function _onStreamDrain (error) {
|
var state = this._spdyState
|
|
state.debug('id=0 _onStreamDrain')
|
|
state.framer.dump()
|
state.framer.unpipe(this.socket)
|
state.framer.resume()
|
|
if (this.socket.destroySoon) {
|
this.socket.destroySoon()
|
}
|
this.emit('close', error)
|
}
|
|
Connection.prototype.end = function end (callback) {
|
var state = this._spdyState
|
|
if (callback) {
|
this.once('close', callback)
|
}
|
this._goaway({
|
lastId: state.stream.lastId.both,
|
code: 'OK',
|
send: true
|
})
|
}
|
|
Connection.prototype.destroyStreams = function destroyStreams (err) {
|
var state = this._spdyState
|
Object.keys(state.stream.map).forEach(function (id) {
|
var stream = state.stream.map[id]
|
|
stream.destroy()
|
if (err) {
|
stream.emit('error', err)
|
}
|
})
|
}
|
|
Connection.prototype.isServer = function isServer () {
|
return this._spdyState.isServer
|
}
|
|
Connection.prototype.getXForwardedFor = function getXForwardFor () {
|
return this._spdyState.xForward
|
}
|
|
Connection.prototype.sendXForwardedFor = function sendXForwardedFor (host) {
|
var state = this._spdyState
|
if (state.version !== null) {
|
state.framer.xForwardedFor({ host: host })
|
} else {
|
state.xForward = host
|
}
|
}
|
|
Connection.prototype.pushPromise = function pushPromise (parent, uri, callback) {
|
var state = this._spdyState
|
|
var stream = this._createStream({
|
request: false,
|
parent: parent,
|
method: uri.method,
|
path: uri.path,
|
host: uri.host,
|
priority: uri.priority,
|
headers: uri.headers,
|
readable: false
|
})
|
|
var err
|
|
// TODO(indutny): deduplicate this logic somehow
|
if (this._isGoaway(stream.id)) {
|
err = new Error('Can\'t send PUSH_PROMISE after GOAWAY')
|
|
process.nextTick(function () {
|
if (callback) {
|
callback(err)
|
} else {
|
stream.emit('error', err)
|
}
|
})
|
return stream
|
}
|
|
if (uri.push && !state.acceptPush) {
|
err = new Error(
|
'Can\'t send PUSH_PROMISE, other side won\'t accept it')
|
process.nextTick(function () {
|
if (callback) { callback(err) } else {
|
stream.emit('error', err)
|
}
|
})
|
return stream
|
}
|
|
stream._sendPush(uri.status, uri.response, function (err) {
|
if (!callback) {
|
if (err) {
|
stream.emit('error', err)
|
}
|
return
|
}
|
|
if (err) { return callback(err) }
|
callback(null, stream)
|
})
|
|
return stream
|
}
|
|
Connection.prototype.setTimeout = function setTimeout (delay, callback) {
|
var state = this._spdyState
|
|
state.timeout.set(delay, callback)
|
}
|
