Network protocol in Node.js

I'd like to know how people approach building network protocols in asynchronous environments such as node.js.

I have implementation of one such protocol, but code seems to complicated for such simple protocol. This is just an exercise. I know I should not implement encryption protocol myself.

I have decent experience with node.js but not with buffers and streams.

• Are there any best practices when it comes to using buffers and streams to implement protocols in this asynchronous world? Any design patterns?
• How can I make this more elegant, more reusable, more flexible? Is my class BufferFiller OK?

Server side:

const net = require('net');
crypto = require("crypto");

const algorithm = "aes-256-cbc";
const host = process.env.HOST || 'localhost';
const port = process.env.PORT || 6000;

class BufferFiller {
    constructor(inputStream) {
        this.inputStream = inputStream;
        this.lastChunkOffset = -1;
        this.lastChunk = null;
    }

    fill(buffer) {
        let bufferFillLength = 0;
        if (this.lastChunk) {
            const copyLen = Math.min(buffer.length - bufferFillLength, this.lastChunk.length - this.lastChunkOffset);
            this.lastChunk.copy(buffer, bufferFillLength, this.lastChunkOffset, this.lastChunkOffset + copyLen);
            bufferFillLength += copyLen;
            this.lastChunkOffset += copyLen;
        }
        if (this.lastChunk && this.lastChunkOffset === this.lastChunk.length) {
            this.lastChunkOffset = -1;
            this.lastChunk = null;
        }
        if (bufferFillLength === buffer.length) {
            return new Promise(resolve => resolve(this));
        }

        return new Promise((resolve, reject) => {
            const onData = chunk => {
                let chunkOffset = 0;
                const copyLen = Math.min(buffer.length - bufferFillLength, chunk.length - chunkOffset);
                chunk.copy(buffer, bufferFillLength, chunkOffset, chunkOffset + copyLen);
                bufferFillLength += copyLen;
                chunkOffset += copyLen;

                if (chunkOffset < chunk.length) {
                    this.lastChunkOffset = chunkOffset;
                    this.lastChunk = chunk;
                }

                if (bufferFillLength === buffer.length) {
                    this.inputStream.removeListener('data', onData);
                    this.inputStream.removeListener('error', onError);
                    resolve(this);
                }
            };
            const onError = err => {
                this.inputStream.removeListener('data', onData);
                this.inputStream.removeListener('error', onError);
                reject(err);
            };
            this.inputStream.on('data', onData);
            this.inputStream.on('error', onError);
        });
    }

    getRemainingBuffer() {
        if (this.lastChunk === null) {
            return Buffer.alloc(0);
        }
        const result = Buffer.alloc(this.lastChunk.length - this.lastChunkOffset);
        this.lastChunk.copy(result, 0, this.lastChunkOffset);
        return result;
    }
};

function exchangeKeysAndPipeThroughDecipher(outputStream, inputStream, nextStream) {
    const iv = Buffer.alloc(16, 0);

    const alice = crypto.createDiffieHellmanGroup('modp14');
    const alicePublicKey = alice.generateKeys();
    const aliceGenerator = alice.getGenerator();
    const alicePrime = alice.getPrime();
    const buf = Buffer.allocUnsafe(4);
    buf.writeUInt32BE(aliceGenerator.length, 0);

    outputStream.write(buf);
    outputStream.write(aliceGenerator);
    outputStream.write(alicePrime);
    outputStream.write(alicePublicKey);

    const bobPublicKey = Buffer.alloc(256);

    new BufferFiller(inputStream)
        .fill(bobPublicKey)
        .then(filler => {
            const symKey = alice.computeSecret(bobPublicKey).slice(-32);
            const decipher = crypto.createDecipheriv(algorithm, symKey, iv);
            decipher.setAutoPadding(false);
            decipher.write(filler.getRemainingBuffer());
            inputStream.pipe(decipher).pipe(nextStream);
        });
};

async function onClientConnected(sock) {
    exchangeKeysAndPipeThroughDecipher(sock, sock, process.stdout);
};

// Create Server instance
var server = net.createServer(onClientConnected);

server.listen(port, host, function() {
    console.log('server listening on %j', server.address());
});

Client side:

const net = require('net');
const crypto = require('crypto');

const algorithm = "aes-256-cbc";
const host = process.env.HOST || 'localhost';
const port = process.env.PORT || 6000;
const client = new net.Socket();

class BufferFiller {
    constructor(inputStream) {
        this.inputStream = inputStream;
        this.lastChunkOffset = -1;
        this.lastChunk = null;
    }

    fill(buffer) {
        let bufferFillLength = 0;
        if (this.lastChunk) {
            const copyLen = Math.min(buffer.length - bufferFillLength, this.lastChunk.length - this.lastChunkOffset);
            this.lastChunk.copy(buffer, bufferFillLength, this.lastChunkOffset, this.lastChunkOffset + copyLen);
            bufferFillLength += copyLen;
            this.lastChunkOffset += copyLen;
        }
        if (this.lastChunk && this.lastChunkOffset === this.lastChunk.length) {
            this.lastChunkOffset = -1;
            this.lastChunk = null;
        }
        if (bufferFillLength === buffer.length) {
            return new Promise(resolve => resolve(this));
        }

        return new Promise((resolve, reject) => {
            const onData = chunk => {
                let chunkOffset = 0;
                const copyLen = Math.min(buffer.length - bufferFillLength, chunk.length - chunkOffset);
                chunk.copy(buffer, bufferFillLength, chunkOffset, chunkOffset + copyLen);
                bufferFillLength += copyLen;
                chunkOffset += copyLen;

                if (chunkOffset < chunk.length) {
                    this.lastChunkOffset = chunkOffset;
                    this.lastChunk = chunk;
                }

                if (bufferFillLength === buffer.length) {
                    this.inputStream.removeListener('data', onData);
                    this.inputStream.removeListener('error', onError);
                    resolve(this);
                }
            };
            const onError = err => {
                this.inputStream.removeListener('data', onData);
                this.inputStream.removeListener('error', onError);
                reject(err);
            };
            this.inputStream.on('data', onData);
            this.inputStream.on('error', onError);
        });
    }

    getRemainingBuffer() {
        if (this.lastChunk === null) {
            return Buffer.alloc(0);
        }
        const result = Buffer.alloc(this.lastChunk.length - this.lastChunkOffset);
        this.lastChunk.copy(result, 0, this.lastChunkOffset);
        return result;
    }
};

client.connect(port, host, function() {
    console.log('Client connected to: ' + host + ':' + port);
    const iv = Buffer.alloc(16, 0);
    const aliceGenLen = Buffer.allocUnsafe(4);
    let aliceGen;
    const alicePrime = Buffer.allocUnsafe(256);
    const alicePublicKey = Buffer.allocUnsafe(256);

    new BufferFiller(client)
        .fill(aliceGenLen)
        .then(filler => {
            const genLen = aliceGenLen.readUInt32BE();
            aliceGen = Buffer.allocUnsafe(genLen);
            return filler.fill(aliceGen);
        })
        .then(filler => filler.fill(alicePrime))
        .then(filler => filler.fill(alicePublicKey))
        .then(filler => {
            const bob = crypto.createDiffieHellman(alicePrime, aliceGen);
            const bobPublicKey = bob.generateKeys();
            client.write(bobPublicKey);
            const symKey = bob.computeSecret(alicePublicKey);
            return symKey.slice(-32);
        })
        .then(symKey => {
            const cipher = crypto.createCipheriv(algorithm, symKey, iv);
            cipher.setAutoPadding(true);
            process.stdin.pipe(cipher).pipe(client);
        });
});

client.on('close', function() {
    console.log('Client closed');
});

client.on('error', function(err) {
    console.error(err);
});

I find implementing protocols in asynchronous world particularly difficult, when you can control how much data you receive from underling buffers, nor when ('data' event triggers with what ever chunk size it wants and when ever it wants).

P.S. I'm stuck with node v8.x, which supports async/await, but I didn't want to hide asynchronicity, on purpose.