index.js

const API_PORT = 14265;
const DELAY = 5; // in miliseconds

// Necessary libraries
const http = require('http');
const pool = new http.Agent({ maxSockets: 64 });
const zmq = require('zeromq');
const sock = zmq.socket('pub');
const Converter = require('./lib/converter.js'); // from iota.lib.js

// Classes
class MessageQ {
	constructor() {
		this.chunks = new Array();
	}
	
	append(chunk) {
		this.chunks.push(chunk);
		return chunk;
	}

	get string() {
		return this.chunks.join(' ');
	}
}

class Tx extends MessageQ {
	// ref:
	//     https://github.com/iotaledger/iri/blob/master/src/main/java/com/iota/iri/storage/ZmqPublishProvider.java#L68
	//     https://github.com/iotaledger/iota.lib.js/blob/develop/lib/utils/utils.js#L169
	constructor(hash, trytes) {
		super();
		const transactionTrits = Converter.trits(trytes);
		this.topic = this.append('tx');
		this.hash = this.append(hash);
		this.address = this.append(trytes.slice(2187, 2268));
		this.value = this.append(Converter.value(transactionTrits.slice(6804, 6837)));
		this.obsoleteTag = this.append(trytes.slice(2295, 2322));
		this.timestamp = this.append(Converter.value(transactionTrits.slice(6966, 6993)));
		this.currentIndex = this.append(Converter.value(transactionTrits.slice(6993, 7020)));
		this.lastIndex = this.append(Converter.value(transactionTrits.slice(7020, 7047)));
		this.bundle = this.append(trytes.slice(2349, 2430));
		this.trunk = this.append(trytes.slice(2430, 2511));
		this.branch = this.append(trytes.slice(2511, 2592));
		this.arrivalDate = this.append(Date.now());
		this.tag = this.append(trytes.slice(2592, 2619));
	}
}

class Tx_trytes extends MessageQ {
	// ref:
	//     https://github.com/iotaledger/iri/blob/master/src/main/java/com/iota/iri/storage/ZmqPublishProvider.java#L93
	constructor (hash, trytes) {
		super();
		this.topic = this.append('tx_trytes');
		this.trytes = this.append(trytes);
		this.hash = this.append(hash);
	}
}

class Sn extends MessageQ {
	// ref:
	//     https://github.com/iotaledger/iri/blob/master/src/main/java/com/iota/iri/LedgerValidator.java#L150
	constructor (milestone, txs) {
		super();
		if (Array.isArray(txs) || txs.length > 5) {
			throw new Error('Confirmed transactions should come in an array of 5');
		}
		this.topic = this.append('sn');
		this.milestone = this.append(milestone);
		this.txs = txs;
		this.txs.forEach(tx => {
			this.append(tx);
		});
	}
}

// Functions
const request = command => {
	return new Promise((resolve, reject) => {
		try {
			const req = http.request({
				protocol: 'http:',
				hostname: 'localhost',
				port: API_PORT,
				method: 'POST',
				agent: pool,
				headers: {
					'Content-Type': 'application/json',
					'X-IOTA-API-Version': 1,
					'Content-Length': Buffer.byteLength(JSON.stringify(command))
				},
			}, res => {
				let data = '';

				res.on('data', chunk => {
					data += chunk;
				});

				res.on('end', () => {
					const body = JSON.parse(data);
					if ('error' in body) {
						throw new Error(body.error);
					}
					setTimeout(resolve, DELAY, body);
				});
			});

			req.on('error', err => {
				throw err;
			});

			req.write(JSON.stringify(command));
			req.end();
		} catch (err) {
			console.error(err);
			reject(err);
		}
	});
};

const getTrytes = async txhash => {
	let { trytes } = await request({ command: 'getTrytes', hashes: [txhash] });
	return trytes[0];
};

const publish = messages => {
	messages.forEach(msg => {
		console.log(msg.string);
		sock.send(msg.string);
	});
};

const traverseForward = async txhash => {
	const { hashes } = await request({ command: 'findTransactions', approvees: [txhash] });
	hashes.forEach(async tx => {
		if (!seenForward.includes(tx)) {
			if (!db.includes(tx)) {
				const trytes = await getTrytes(tx);
				const msgs = new Array();
				msgs.push(new Tx(tx, trytes));
				msgs.push(new Tx_trytes(tx, trytes));
				publish(msgs);
				save(tx);
			}
			queueForward.push(tx);
			seenForward.push(tx);
		}
	});
	queueForward.shift();
};

const traverseBackward = async txhash => {
	const trytes = await getTrytes(txhash);

	if (!db.includes(txhash)) {
		const trytes = await getTrytes(txhash);
		const msgs = new Array();
		msgs.push(new Tx(txhash, trytes));
		msgs.push(new Tx_trytes(txhash, trytes));
		publish(msgs);
		db.push(txhash);
	}

	const parents = [ trytes.slice(2430, 2511), trytes.slice(2511, 2592) ];
	parents.forEach(async tx => {
		if (!seenBackward.includes(tx)) {
			if (tx === '9'.repeat(81)) {
				// reached the genesis. traversing forwards again...
				queueForward.push(txhash);
			} else {
				queueBackward.push(tx);
				seenBackward.push(tx);
			}
		}
	});
	queueBackward.shift();
};

// Declerations
const seenBackward = new Array();
const seenForward = new Array();
const queueBackward = new Array();
const queueForward = new Array();
const db = new Array();
let counter = 0;


// Do the thang
(async () => {
	const { latestSolidSubtangleMilestone } = await request({ command: 'getNodeInfo' });
	queueBackward.push(latestSolidSubtangleMilestone);
	
	sock.bindSync('tcp://127.0.0.1:5557');

	// First traverse backwards towards the milestone, publishing confirmed transactions
	while (queueBackward.length > 0) {
		await traverseBackward(queueBackward[0]);
	}

	// Not sure if needed, but delete seen backward transactions object since we're done here
	delete seenBackward;

	// And then from the genesis traverse forward towards the edge, publishing unconfirmed transactions
	while (queueForward.length > 0) {
		await traverseForward(queueForward[0]);
	}

	// Not sure if needed, but delete seen forward transactions object since we're done here
	delete seenForward;
})();