git ssb

1
// Copyright 2012 Iris Couch, all rights reserved.
2
//
3
// Server routines
4
5
require('defaultable')(module,
6
  {
7
  }, function(module, exports, DEFS, require) {
8
9
var net = require('net')
10
var util = require('util')
11
var dgram = require('dgram')
12
var events = require('events')
13
14
var Message = require('./message')
15
var convenient = require('./convenient')
16
17
module.exports = createServer
18
19
function createServer(handler) {
20
  return new Server(handler)
21
}
22
23
function bindUDP (self, type) {
24
  self.udp = dgram.createSocket(type || 'udp4')
25
  self.udp.on('close', function() { self.close() })
26
  self.udp.on('error', function(er) { self.emit('error', er) })
27
  self.udp.on('message', function(msg, rinfo) { self.on_udp(msg, rinfo) })
28
  self.udp.once('listening', function() {
29
    self.listening.udp = true
30
    if(self.listening.tcp)
31
      self.emit('listening')
32
  })
33
}
34
35
function bindTCP (self) {
36
  self.tcp = net.createServer()
37
  self.tcp.on('close', function() { self.close() })
38
  self.tcp.on('error', function(er) { self.emit('error', er) })
39
  self.tcp.on('connection', function(connection) { self.on_tcp_connection(connection) })
40
  self.tcp.once('listening', function() {
41
    self.listening.tcp = true
42
    if(self.listening.udp)
43
      self.emit('listening')
44
  })
45
}
46
47
util.inherits(Server, events.EventEmitter)
48
function Server (handler) {
49
  var self = this
50
  events.EventEmitter.call(self)
51
52
  self.log = console
53
  self.zones = {}
54
  self.listening = {'tcp':false, 'udp':false}
55
56
  if(handler) self.on('request', handler)
57
}
58
59
Server.prototype.zone = function(zone, server, admin, serial, refresh, retry, expire, ttl) {
60
  var self = this
61
    , record = zone
62
63
  if(typeof record != 'object')
64
    record = { 'class': 'IN'
65
             , 'type' : 'SOA'
66
             , 'name' : zone
67
             , 'data' : { 'mname': server
68
                        , 'rname': admin
69
                        , 'serial': convenient.serial(serial)
70
                        , 'refresh': convenient.seconds(refresh)
71
                        , 'retry'  : convenient.seconds(retry)
72
                        , 'expire' : convenient.seconds(expire)
73
                        , 'ttl'    : convenient.seconds(ttl || 0)
74
                        }
75
             }
76
77
  self.zones[record.name] = record
78
  return self
79
}
80
81
Server.prototype.listen = function(port, ip, callback) {
82
  var self = this
83
84
  if(typeof ip === 'function') {
85
    callback = ip
86
    ip = null
87
  }
88
89
  self.port = port
90
  self.ip   = ip || '0.0.0.0'
91
92
  if(typeof callback === 'function')
93
    self.on('listening', callback)
94
95
  bindUDP(self, net.isIPv6(self.ip)?'udp6':'udp4');
96
  self.udp.bind(port, ip)
97
98
  bindTCP(self);
99
  self.tcp.listen(port, ip)
100
101
  return self
102
}
103
104
Server.prototype.close = function() {
105
  var self = this
106
107
  if(self.udp._receiving)
108
    self.udp.close()
109
110
  if(self.tcp._handle)
111
    self.tcp.close(function() {
112
      self.emit('close')
113
    })
114
}
115
116
Server.prototype.unref = function() {
117
  this.udp.unref()
118
  this.tcp.unref()
119
}
120
121
Server.prototype.ref = function() {
122
  this.udp.ref()
123
  this.tcp.ref()
124
}
125
126
Server.prototype.on_tcp_connection = function(connection) {
127
  var self = this
128
129
  var length = null
130
    , bufs = []
131
132
  connection.type = 'tcp'
133
  connection.server = self
134
135
  connection.on('data', function(data) {
136
    bufs.push(data)
137
    var bytes_received = bufs.reduce(function(state, buf) { return state + buf.length }, 0)
138
139
    if(length === null && bytes_received >= 2) {
140
      var so_far = Buffer.concat(bufs) // Flatten them all together, it's probably not much data.
141
      length = so_far.readUInt16BE(0)
142
      bufs = [ so_far.slice(2) ]
143
    }
144
145
    if(length !== null && bytes_received == 2 + length) {
146
      // All of the data (plus the 2-byte length prefix) is received.
147
      try {
148
        var data = Buffer.concat(bufs)
149
          , req = new Request(data, connection)
150
          , res = new Response(data, connection)
151
152
        self.emit('request', req, res)
153
      } catch(err) {
154
        self.emit('error', 'Error processing request', err, connection)
155
      }
156
    }
157
  })
158
}
159
160
Server.prototype.on_udp = function(data, rinfo) {
161
  var self = this
162
163
  // Provide something that feels like a net.Socket, which in turn feels like the http API.
164
  var connection = { 'type'         : self.udp.type
165
                   , 'remoteAddress': rinfo.address
166
                   , 'remotePort'   : rinfo.port
167
                   , 'server'       : self
168
                   , 'send'         : function() { self.udp.send.apply(self.udp, arguments) }
169
                   , 'destroy'      : function() {}
170
                   , 'end'          : function() {}
171
                   }
172
173
  try {
174
    var req = new Request(data, connection)
175
      , res = new Response(data, connection)
176
177
    self.emit('request', req, res)
178
  } catch (err) {
179
    self.emit('error', 'Error processing request', err, connection)
180
  }
181
}
182
183
util.inherits(Request, Message)
184
function Request (data, connection) {
185
  var self = this
186
  Message.call(self, data)
187
188
  self.connection = connection
189
}
190
191
Request.prototype.toJSON = function() {
192
  var self = this
193
  var obj = {}
194
  Object.keys(self).forEach(function(key) {
195
    if(key != 'connection')
196
      obj[key] = self[key]
197
  })
198
  return obj
199
}
200
201
util.inherits(Response, Message)
202
function Response (data, connection) {
203
  var self = this
204
  Message.call(self, data)
205
206
  self.question   = self.question   || []
207
  self.answer     = self.answer     || []
208
  self.authority  = self.authority  || []
209
  self.additional = self.additional || []
210
211
  self.connection = connection
212
213
  convenient.init_response(self)
214
}
215
216
Response.prototype.toJSON = Request.prototype.toJSON
217
218
var udp4Or6 = function (t) { return ['udp4', 'udp6'].indexOf(t) !== -1 };
219
220
Response.prototype.end = function(value) {
221
  var self = this
222
223
  var msg = convenient.final_response(self, value)
224
    , data = msg ? msg.toBinary() : self.toBinary()
225
226
  if(udp4Or6(self.connection.type)&& data.length > 512)
227
    return self.emit('error', 'UDP responses greater than 512 bytes not yet implemented')
228
229
  else if(udp4Or6(self.connection.type))
230
    self.connection.send(data, 0, data.length, self.connection.remotePort, self.connection.remoteAddress, function(er) {
231
      if(er)
232
        self.emit('error', er)
233
    })
234
235
  else if(self.connection.type == 'tcp') {
236
    // Add the data length prefix.
237
    var length = data.length
238
    data = Buffer.concat([ new Buffer([length >> 8, length & 255]), data ])
239
240
    try {
241
      self.connection.end(data, function(er) {
242
        if(er)
243
          self.emit('error', er)
244
      })
245
    } catch(er) {
246
      self.emit('error', er)
247
    }
248
  }
249
250
  else
251
    self.emit('error', new Error('Unknown connection type: ' + self.connection.type))
252
}
253
254
255
}) // defaultable
256