{"size": 1185, "ext": "js", "max_stars_repo_path": "lib/receiver.js", "max_stars_repo_name": "derhuerst/lossy-state-sync-stream", "max_stars_repo_stars_event_min_datetime": "2020-05-29T21:03:03.000Z", "max_stars_repo_stars_event_max_datetime": "2020-05-29T21:03:03.000Z", "max_issues_repo_path": "lib/receiver.js", "max_issues_repo_name": "derhuerst/lossy-state-sync-stream", "max_issues_repo_issues_event_min_datetime": "2020-05-29T18:17:36.000Z", "max_issues_repo_issues_event_max_datetime": "2020-05-30T22:54:04.000Z", "max_forks_repo_path": "lib/receiver.js", "max_forks_repo_name": "derhuerst/lossy-state-sync-stream", "max_forks_repo_forks_event_min_datetime": null, "max_forks_repo_forks_event_max_datetime": null}
{"max_stars_count": 1, "max_issues_count": 3, "max_forks_count": null, "avg_line_length": 22.7884615385, "max_line_length": 70, "alphanum_fraction": 0.6776371308}
'use strict'

const debug = require('debug')('lossy-state-sync-stream:receiver')
const {Duplex} = require('stream')
const {encode: encodeVarint, decode: decodeVarint} = require('varint')
const {encodeMsg, decodeMsg} = require('./encoding')

const createReceiver = () => {
	let highestSeenSeq = -Infinity
	let highestAckedSeq = -Infinity

	const handleSenderMessage = (msg, _, cb) => {
		const t = Date.now()
		const {
			seq,
			payload,
		} = decodeMsg(msg)

		if (seq <= highestSeenSeq) return cb() // ignore msg
		highestSeenSeq = seq

		if (seq >= highestAckedSeq) {
			debug('sending ack for', seq)
			receiver.push(encodeMsg(seq))
			highestAckedSeq = seq
		}

		// no payload a.k.a. ping-only msg
		if (payload.length === 0) return cb()

		debug('received', seq, payload)
		receiver.emit('message', payload)
		cb()
	}

	const receiver = new Duplex({
		// readable a.k.a. receiver -> sender
		readableObjectMode: true,
		readableHighWaterMark: 1,
		read: () => {},
		// writable a.k.a. receiver -> sender
		writableObjectMode: true,
		writableHighWaterMark: 1,
		write: handleSenderMessage,
	})
	receiver.senderAddress = null

	return receiver
}

module.exports = createReceiver
