Connectable protocol

[hamiltop]

First pass at a protocol is

streamz/lib/streamz/merge.ex

Line 93 in 6c416d0

defprotocol Mergeable do

The goal is to make this more generic (not just specific to merge). Initial thought is:

defprotocol Connectable do
  def connect(source, target)
  def disconnect(connection)
end

I think that connect should return some sort of connection. disconnect will accept that as an input. This will contain the protocol so implementations can do whatever they want.

Potential uses:

Stream.Merge would connect each of the substreams to the target, allowing us to collapse nested merges.

GenEvent.Streams could be merged without an intermediate process.

Stream.Interval could call :timer.send_interval rather than spawn a new process. disconnect would cancel the timer.

and more...

[hamiltop]

This is an odd protocol because it depends heavily on side effects. I don't think we have any other protocols that depend on side effects right now.

Documenting this is going to be hard.

Making it easy to implement connectable is going to take some thought.

[hamiltop]

First pass at a Connectable Protocol.

https://github.com/hamiltop/streamz/blob/connectable_protocol/lib/streamz/connectable.ex

For GenEvent, it does not remove handlers.

Surprisingly, the performance of GenEvent Streams being merged dropped 5% when sending directly. That's really weird. I'm surprised it's faster to have an intermediate process.

[josevalim]

That looks great to me. I think you should be able to disconnect a gen event stream by calling: GenEvent.remove_handler(manager, {GenEvent.Stream, ref}, []) where ref is the reference you get on "connect".

[josevalim]

We are also changing the protocol to not rely on :gen_event_EXIT but rather on a stop message (which will be sync).

[hamiltop]

That should work.

Connect will differ from a normal GenEvent.Stream in that there will be no monitoring. This sort of ties into the whole idea of error handling.

We need to figure out a bit about exit cases. In the case of merge, we want to know when all the streams have exited so we can finish the stream. Will monitoring be desired in every circumstance? Should that be a part of Connectable?

Documenting this is going to be tricky, because there are a lot of side effects expected.

[josevalim]

Connect will differ from a normal GenEvent.Stream in that there will be no monitoring. This sort of ties into the whole idea of error handling.

The monitoring or linking is completely up to you, yes.

I have noticed that your example uses :done to notify the merger process that a stream is done but that is not part of the protocol (we use :done to notify we will no longer be consuming). So we need a way to fix that.

I was discussing with @fishcakez that the best option is to add a :stop message to the protocol. In the case of merger, you will likely need to make a distinction between a stop from one of the children processes being merged and an external request to stop, but I think we would be fine?

[hamiltop]

Yeah, I didn't update mine because I knew you guys were coming up with something else.

Who is the monitoring or linking up to? The caller (merge) or the implementation of Connectable?

if it's up to the caller, then we need to expose the pid explicitly. Right now the return from connect is an opaque value that is passed to the disconnect call. It could be a pid, it could be something else.

Sort of hard to make the call without a variety of callers of connect

[josevalim]

I think we could return {pid, ref} and people can link and monitor afterwards as they prefer.

[hamiltop]

That's along the lines of what I was thinking. I'm going to try my hand at
documenting the protocol, outlining expectations and such. Stay tuned.

On Tue, Aug 26, 2014 at 8:42 AM, José Valim notifications@github.com
wrote:

I think we could return {pid, ref} and people can link and monitor
afterwards as they prefer.

—
Reply to this email directly or view it on GitHub
#19 (comment).