add documentation to event-emitter

[coolaj86]

We have here a bunch of code, but no documentation: https://github.com/neon-bindings/examples/tree/master/event-emitter

It would be nice to know:

• How to build it? (probably something like npm install neon; neon build)
• How to use it? (probably something like var EvEm = require("event-emitter"); ...)

It looks like there's a lot of code there. Could we perhaps pare it down to just an event emitter example and split the polling and other stuff into a second example?

It's probably perfect as-is for a quick reference for those that are familiar with Rust and with neon and I may be very much in the minority, but I'm looking at neon as an excuse to become more familiar with Rust and although I know that what I need is right in front of me, I don't know enough to easily pick it out yet.

[coolaj86]

I spent some time going through this and either I'm completely misunderstanding what it does, or this isn't an event emitter at all.

It looks like it's simply exposing an async method poll() which puts an event onto a schedule and then calls the callback which was handed to poll(), and loops to call poll() again with a new function.

Is there anyway to adapt this so that the callback is called multiple times like an event emitter would be (i.e. getting data from a network socket) rather than having the upper level JS telling the lower level Rust to call the upper level JS?

Or how would I get data from a network socket on a regular basis? The data is coming in whether or not the JavaScript asks for it, which is what normally makes callbacks a nice pattern.

Ultimately I suppose that I do want a "pull" stream rather than a "push" stream, and maybe that's what this is really demonstrating, but I'd like to see how to have a looping thread in Rust fire a callback every 500ms. Here it looks like if I do a setTimeout(poll, 1500), then the count doesn't continue to increment. Instead it sits and waits for the upper JS to tell it to continue its would-be background process.

[amilajack]

This example was taken from here. @kjvalencik would be the best to answer this. See neon-bindings/neon#386 (comment) for some details on it.

[kjvalencik]

@coolaj86 Hi! Sorry for the lack of documentation. This is something that I threw together quickly to provide as an example for someone in our slack channel. It's not as complete as a proper example should be.

Firstly, I should mention that this entire example is a hack around a missing feature in Neon. Namely, there is no way to schedule code to execute on the main JS thread. This is something that is being considered.

neon-bindings/rfcs#25
neon-bindings/neon#375

Next, this example has a lot of code that isn't relevant to the event emitter example. The part that is relevant to the EventEmitter:

examples/event-emitter/native/src/lib.rs

Lines 158 to 262 in 84b125c

	pub enum Event {
	Tick { count: f64 },
	}
	fn event_thread(rx: mpsc::Receiver<()>) -> mpsc::Receiver<Event> {
	let (tx, events) = mpsc::channel();
	thread::spawn(move || {
	let mut count = 0.0;
	loop {
	thread::sleep(Duration::from_millis(500));
	match rx.try_recv() {
	Ok(_) | Err(TryRecvError::Disconnected) => {
	break;
	}
	Err(TryRecvError::Empty) => {}
	}
	tx.send(Event::Tick { count }).expect("Send failed");
	count += 1.0;
	}
	});
	events
	}
	pub struct EventEmitter {
	events: Arc<Mutex<mpsc::Receiver<Event>>>,
	shutdown: mpsc::Sender<()>,
	}
	pub struct EventEmitterTask(Arc<Mutex<mpsc::Receiver<Event>>>);
	impl Task for EventEmitterTask {
	type Output = Event;
	type Error = String;
	type JsEvent = JsObject;
	fn perform(&self) -> Result<Self::Output, Self::Error> {
	let rx = self
	.0
	.lock()
	.map_err(|_| "Could not obtain lock on receiver".to_string())?;
	rx.recv().map_err(|_| "Failed to receive event".to_string())
	}
	fn complete(
	self,
	mut cx: TaskContext,
	event: Result<Self::Output, Self::Error>,
	) -> JsResult<Self::JsEvent> {
	let event = event.or_else(|err| cx.throw_error(&err.to_string()))?;
	let o = cx.empty_object();
	match event {
	Event::Tick { count } => {
	let event_name = cx.string("tick");
	let event_count = cx.number(count);
	o.set(&mut cx, "event", event_name)?;
	o.set(&mut cx, "count", event_count)?;
	}
	}
	Ok(o)
	}
	}
	declare_types! {
	pub class JsEventEmitter for EventEmitter {
	init(_) {
	let (shutdown, shutdown_rx) = mpsc::channel();
	let rx = event_thread(shutdown_rx);
	Ok(EventEmitter {
	events: Arc::new(Mutex::new(rx)),
	shutdown,
	})
	}
	method poll(mut cx) {
	let cb = cx.argument::<JsFunction>(0)?;
	let this = cx.this();
	let events = cx.borrow(&this, |emitter| Arc::clone(&emitter.events));
	let emitter = EventEmitterTask(events);
	emitter.schedule(cb);
	Ok(JsUndefined::new().upcast())
	}
	method shutdown(mut cx) {
	let this = cx.this();
	cx.borrow(&this, |emitter| emitter.shutdown.send(()))
	.or_else(|err| cx.throw_error(&err.to_string()))?;
	Ok(JsUndefined::new().upcast())
	}
	}
	}

High level, since a Rust thread can't call into JS, something from JS needs to call into Rust. This is what the poll API provides.

The event_thread method simulates a long running process that sends data to a Channel. In this example, the channel is unbounded and will continue to accumulate data if it's not read from. In your example, this would be a Rust thread reading from a network socket and sending the bytes on the channel.

examples/event-emitter/native/src/lib.rs

Line 162 in 84b125c

fn event_thread(rx: mpsc::Receiver<()>) -> mpsc::Receiver<Event> {

The poll method exposed from Neon reads from the channel. It uses a blocking read that waits indefinitely for data to be available. This is why it's an asynchronous method. We don't want to block the event loop while waiting for data. However, it's still taking a LIB_UV thread, so ideally it would yield occasionally anyway.

The data that is being written to the channel looks like { event: string, ...data }. The EventEmitter JS code is glue code to convert the poll interface into something that is more typical JS. It polls in a loop, taking data each time and emitting an event with this.emit(event, data).

examples/event-emitter/lib/emitter.js

Line 20 in 84b125c

.then(({ event, ...data }) => this.emit(event, data))

With the glue code out of the way, a consumer of the event emitter can subscribe to events.

examples/event-emitter/lib/emitter.js

Line 36 in 84b125c

emitter.on('tick', ({ count }) => console.log(count));

In your use case, it might make more sense to adapt a stream interface rather than an EventEmitter.

I hope this helps!

[coolaj86]

Wow! Thanks so much.

[msansoni]

Thanks for everything on this example! However I'm struggling with the example erroring out:

>domain.js:129
domain.enter();
TypeError: domain.enter is not a function

It appears to be as a result of the attempted task scheduling in the rust module:

examples/event-emitter/native/src/lib.rs

Line 180 in f76cd98

emitter.schedule(cb);