tcp-raw.js

/* does effectively this, but without all that fancy
 * stream mumbo jumbo:

var net = require('net');

var server = net.createServer(function(socket) {
  socket.pipe(socket);
});

server.listen(PORT, function() {
  var client = net.connect(PORT, function() {
    client.on('data', function(chunk) {
      console.log('CLIENT %d', chunk.length);
    });
    var n = 16;
    client.write(new Buffer(bufSize), function w() {
      if (--n > 0)
        client.write(new Buffer(bufSize), w);
      else
        server.close();
    });
  });
});

*/


process.on('uncaughtException', function(e) {
  console.error(e.stack || e.message);
  console.error('errno = %j', global.errno);
  process.exit(global.errno || 1);
});

var PORT = 1337;
var TCP = process.binding('tcp_wrap').TCP;
var writeWaterMark = 1024 * 16;
var bufSize = 1024 * 1024;
var bufCount = 16;
var totalBuf = bufSize * bufCount;


//
// SERVER
//

var serverHandle = new TCP();

// server.listen(PORT)
var r = serverHandle.bind('127.0.0.1', PORT);
if (r) {
  serverHandle.close();
  throw new Error('bind');
}

serverHandle.onconnection = onconnection;

var r = serverHandle.listen(511);
if (r) {
  serverHandle.close();
  throw new Error('listen');
}

console.log('listening');

function onconnection(clientHandle) {
  console.error('onconnection');

  if (!clientHandle)
    throw new Error('connect');

  clientHandle.server = true;

  // here goes the actual connection handler stuff
  connectionHandler(clientHandle);
}

function connectionHandler(clientHandle, fn) {
  clientHandle.onread = onread(selfPipe)
  clientHandle.pendingWrites = 0;
  clientHandle.readStart();
}

// socket.pipe(socket);
function selfPipe(handle, chunk) {
  var writeReq = handle.writeBuffer(chunk);
  if (!writeReq || typeof writeReq !== 'object')
    throw new Error('write');

  writeReq.oncomplete = afterWrite(selfPipeAfterWrite);

  handle.pendingWrites++;

  handle.readStop();
}
function selfPipeAfterWrite(handle, req) {
  handle.readStart();
}

function onread(fn) { return function(buffer, offset, length) {
  var handle = this;

  if (buffer) {
    if (length === 0)
      return;

    var d = handle.server ? 'SERVER' : 'CLIENT';

    if (!handle.totalRead)
      handle.totalRead = 0;

    handle.totalRead += length;

    console.error('%s ondata %j %j', d, length, handle.totalRead, totalBuf);
    var chunk = buffer.slice(offset, offset + length);
    fn(handle, chunk);

  } else if (errno === 'EOF') {
    console.error('%s EOF', handle.server ? 'SERVER' : 'CLIENT');
    destroySoon(handle);

  } else if (errno === 'ECONNRESET') {
    console.error('%s ECONNRESET', handle.server ? 'SERVER' : 'CLIENT');
    destroy(handle);

  } else {
    throw new Error('read');
  }
}}

function afterWrite(fn) { return function(status, handle, req) {
  if (status)
    throw new Error('write');

  console.error('afterWrite');

  handle.pendingWrites--;

  if (handle.shutdowning)
    return shutdownSoon(handle);

  if (handle.destroying)
    return destroySoon(handle);

  // done with the write, start reading again.
  fn(handle, req);
}}

function destroySoon(handle) {
  handle.destroying = true;
  if (handle.pendingWrites === 0)
    return destroy(handle);
}

function destroy(handle) {
  handle.close();
  handle.onread = function() {};

  if (handle.onclose)
    handle.onclose();
}



//
// CLIENT
//

var requestHandle = new TCP();
requestHandle.onread = onread(function(handle, chunk) {
  console.error('CLIENT', chunk.length);
});

requestHandle.onclose = function() {
  console.error('CLIENT close');
  serverHandle.close();
};

requestHandle.pendingWrites = 0;

var connectReq = requestHandle.connect('127.0.0.1', PORT);
if (!connectReq)
  throw new Error('connect (1)');
connectReq.oncomplete = afterConnect;

function afterConnect(status, handle, req, readable, writable) {
  if (status)
    throw new Error('connect (2)');

  console.error('CLIENT connect r=%j w=%j', readable, writable);

  handle.readStart();

  clientWrite(handle);
}

function clientWrite(handle) {
  // now just write a big thing
  var writeReq = handle.writeBuffer(new Buffer(bufSize));
  writeReq.oncomplete = afterWrite(function(handle, req) {
    console.error('CLIENT finished writing');
    if (-- bufCount > 0)
      clientWrite(handle);
    else
      shutdownSoon(handle);
  });
  requestHandle.pendingWrites++;
}

function shutdownSoon(handle) {
  handle.shutdowning = true;
  if (handle.pendingWrites === 0)
    return shutdown(handle);
}

function shutdown(handle) {
  var shutdownReq = handle.shutdown();
  if (!shutdownReq)
    throw new Error('shutdown (1)');
  shutdownReq.oncomplete = afterShutdown;
}

function afterShutdown(status, handle, req) {
  if (status)
    throw new Error('shutdown (2)');
}