syncpipes.js

#!/usr/bin/env node
// Creates pipes from different ports to a single sync_gateway
// Useful for debugging syncing/conflicts
//
// ❯ node scripts/syncpipes.js
//
// syncpipes> status
// [1]: 3984 => 4984 | 1 connection  | 8 seconds ago
// [2]: 5984 => 4984 | 1 connection  | 7 seconds ago
//
// syncpipes> stop 1
// [1]: 3984 => 4984 | closed        | 11 seconds ago
// [2]: 5984 => 4984 | 1 connection  | 10 seconds ago
//
// syncpipes> start 1
// [1]: 3984 => 4984 | 0 connections | 17 seconds ago
// [2]: 5984 => 4984 | 1 connection  | 16 seconds ago
//
// syncpipes> status
// [1]: 3984 => 4984 | 1 connection  | 2 seconds ago
// [2]: 5984 => 4984 | 3 connections | 2 seconds ago
//
const net = require('net');
const readline = require('readline');

const SG_PORT = 4984;

// Pipes to be created
// Non-zero keys only
const pipes = {
  1: {
    port: 3984,
    server: null
  },
  2: {
    port: 5984,
    server: null
  }
}

// unix timestamp
const ts = () => Math.floor(new Date() / 1000)

// Function to create a server piping from one port to SG_PORT
const createServer = port => {
  const server = net.createServer(socket => {
    const client = new net.Socket();
    client.connect(SG_PORT, 'localhost', () => {
      socket.pipe(client);
      client.pipe(socket);
    });
    client.on('data', () => {
      server.lastActivity = ts()
    });
  });

  // Keep track of connections so we can close them all
  const connections = {}

  server.on('connection', c => {
    const { remoteAddress, remotePort } = c;
    const key = `${remoteAddress}:${remotePort}`;
    connections[key] = c;
    c.on('close', () => {
      delete connections[key];
    });
  });

  // Function to destroy server and its connections
  server.destroy = cb => {
    server.close(cb);
    for (const key in connections) {
      connections[key].destroy();
    }
  };

  server.listen(port);

  return server;
}

// Create servers
for (const key in pipes) {
  const pipe = pipes[key]
  pipe.server = createServer(pipe.port)
}

// Create prompt
const rl = readline.createInterface({
  input: process.stdin,
  output: process.stdout
});

// Clear screen and reset cursor
readline.cursorTo(process.stdout, 0, 0);
readline.clearScreenDown(process.stdout);

rl.setPrompt('\nsyncpipes> ');

rl.prompt();

const lastActivity = server => {
  return `${ts() - server.lastActivity} seconds ago`
}

const COLUMN_WIDTH = 13

const COLORS = {
  red: '31',
  green: '32',
  cyan: '36',
}

// Function for pretty printing status
const printStatus = (done) => {
    Object.entries(pipes).forEach(([i, pipe], index) => {
      const log = (status, color) => {
        const padded = status.padEnd(COLUMN_WIDTH)
        console.log(
          `\x1b[${color}m%s\x1b[0m`,
          `[${i}]: ${pipe.port} => ${SG_PORT} | ${padded} | ${lastActivity(pipe.server)}`
        )
      }
      if (pipe.server) {
        pipe.server.getConnections((err, count) => {
          const isListening = pipe.server.listening;
          log(
            isListening ? `${count} connection${count === 1 ? '' : 's'}` : 'closed',
            isListening ? (count === 0 ? COLORS.cyan : COLORS.green) : COLORS.red
          )
          if (index === Object.keys(pipes).length - 1) {
            done()
          }
        });
      } else if (pipe.server === null) {
        log('uninitialized', COLORS.red)
        done()
      } else {
        log('unknown', COLORS.red)
        done()
      }
    });
}

rl.on('line', line => {
  const cmd = line.trim();

  const matchPipeNumber = startOrStop => {
    const match = cmd.match(new RegExp(`^${startOrStop}\\s(\\d)$`));
    if (match) {
      return match[1];
    }

    return null;
  }

  const startPipeNumber = matchPipeNumber('start');
  const stopPipeNumber = matchPipeNumber('stop');

  if (cmd === 'status') {
    printStatus(() => rl.prompt());
  } else if (startPipeNumber) {
    const pipe = pipes[startPipeNumber];

    if (!pipe) {
      console.log('Pipe not found');
      rl.prompt();
      return;
    }

    const { server, port } = pipe

    server.listen(port);
    printStatus(() => rl.prompt());
  } else if (stopPipeNumber) {
    const pipe = pipes[stopPipeNumber];

    if (!pipe) {
      console.log('Pipe not found');
      rl.prompt();
      return;
    }

    const { server, port } = pipe

    if (server) {
      server.destroy(() => {
        printStatus(() => rl.prompt());
      });
    } else {
      console.log(`No pipe open: ${port}`)
      rl.prompt();
    }
  } else {
    console.log('Invalid command')
    rl.prompt();
  }
}).on('close', () => {
  console.log('\n\nBye!');
  // This will kill the connections/servers anyway
  process.exit(0);
});