Copy Page

You are not logged in. Any changes you make will be marked as anonymous. You may want to Log In if you already have an account. You can also Sign Up for a new account.

<h2>Problems</h2>

<ul>
	<li>If the generating function of a lazy sequence blocks (e.g. for I/O), it blocks all consuming threads</li>
	<li>Clojure's sequence functions cannot be applied to asynchronous events without converting events to blocking sequences</li>
</ul>

<h2>Proposed Solution</h2>

<ul>
	<li>Protocols for generators and consumers of asynchronous events</li>
	<li>Functions which mirror the Clojure sequence API (e.g. map, filter) for &quot;push&quot; events</li>
	<li>Integrate with <a href="/display/design/Scheduled+Events" data-linked-resource-id="1573161" data-linked-resource-type="page" data-linked-resource-default-alias="Scheduled Events" data-base-url="http://dev.clojure.org">Scheduled Events</a></li>
</ul>

<h2>References</h2>

<ul>
	<li><a href="http://msdn.microsoft.com/en-us/devlabs/ee794896.aspx">Microsoft Reactive Extensions for .NET (Rx)</a></li>
	<li><a href="http://okmij.org/ftp/Haskell/Iteratee/IterateeIO-talk-notes.pdf">Haskell Iteratee IO</a></li>
	<li><a href="http://www.cs.brown.edu/~greg/thesis.pdf">Integrating Dataﬂow Evaluation into a Practical Higher-Order Call-by-Value Language</a> by Greg Cooper
	<ul>
		<li>See also Cooper's FrTime in Racket: <a href="http://docs.racket-lang.org/frtime/index.html">FrTime: A Language for Reactive Programs</a></li>
	</ul>
	</li>
</ul>

<h2>Experiments</h2>

<ul>
	<li><a href="https://github.com/stuartsierra/cljque">Cljque</a> by Stuart Sierra
	<ul>
		<li>Protocols &quot;Observable&quot; and &quot;Observer&quot; very similar to Rx</li>
		<li>&quot;Push&quot; sequence API with fns that look like the sequence API, with the same names</li>
		<li>Incomplete, currently-broken integration with Netty</li>
	</ul>
	</li>
	<li><a href="https://github.com/ztellman/lamina">Lamina</a> by Zach Tellman
	<ul>
		<li>Foundation for <a href="https://github.com/ztellman/aleph">Aleph</a></li>
		<li>No direct analogue for first/rest, but event stream equivalents for map/filter/reduce/take</li>
		<li>Based on <a href="https://github.com/ztellman/lamina/wiki/Channels">Channels</a> (previous discussion <a href="/display/design/Channels" data-linked-resource-id="950497" data-linked-resource-type="page" data-linked-resource-default-alias="Channels" data-base-url="http://dev.clojure.org">here</a>)
		<ul>
			<li>Queues to which &quot;handler&quot; functions can subscribe</li>
			<li>Permits both &quot;push&quot; and &quot;pull&quot; on the same data</li>
		</ul>
		</li>
	</ul>
	</li>
	<li><a href="https://github.com/hiredman/die-geister">Die Geister</a> by Kevin Downey</li>
</ul>

<h2>Design Discussion: April 1, 2011</h2>

<h6>Models of Event Handling</h6>

<p>3 models for events: pull (blocking), poll (nonblocking), and push (registered event handlers).  Poll isn't really a model; it always ends up being pull somewhere.  Event handlers usually end up being dead-ends – once a handler is triggered, the event stops.</p>

<h6>Models of Iteration</h6>

<p>Iteration has three operations: 1) is there more stuff? 2) get the stuff; and 3) move.  In Clojure, these correspond to 'seq', 'first', and 'rest'.  ('next' is derived as the 'seq' of 'rest'.)  Java's iterators and .NET's enumerators each conflate 2 out of the 3, but in different combinations.   What are the corresponding operations for asynchronous events?</p>

<h6>Unsubscribing</h6>

<p>In .NET, subscribing to an IObservable returns an IDisposable, which connects to the rest of the .NET resource management infrastructure. In Java, the closest analog is java.io.Closeable, so subscribing to an observable object should return a Closeable.</p>

<h6>Which Thread to Run On</h6>

<p>Event handlers can be invoked on the thread that generated the event or dispatched to another thread (such as the Agent thread pool). However, if the framework forces dispatch to another thread, you have no choice. If execution stays on the thread that generated the event, you always have the option of dispatching to another thread.</p>

<h6>Duality and Naming</h6>

<p>Eric Meyer's talk on Rx emphasizes that IObserver/IObservable is a dual to Enumerator/Enumerable. What is the dual to Clojure seqs? Seqs are nice because they're simple: the body of a seq function boils down to one block of code, a closure. Can asynchronous events be made equally simple? Can you merge the &quot;on event,&quot; &quot;on completed,&quot; and &quot;on error&quot; operations of an observer into one block of code?</p>

<h6>Ordering</h6>

<p>The &quot;pull&quot; model implies ordering; you only deal with one thing at at a time. In the &quot;push&quot; model, do you still have to provide that, or can you send many events at once? Do event handlers need to be reentrant? (The Rx Design Guidelines, sec. 6.7, say that calls to IObserver methods should be serialized so that consumers do not need to deal with concurrency.)</p>

<p>In the &quot;push&quot; model, the order in which events are received and dispatched is non-deterministic. One problem people have with Rx occurs when you receive an event and start a process based on it, assuming you're finished. But then you receive another event from the same source.</p>

<h2>Design Discussion: April 8, 2011</h2>

<p>Reactive Extensions' IObservable isn't really a dual of IEnumerator.  The arguments don't line up, and the fact that Subscribe returns an IDisposable complicates the model.  In the Observable model, the &quot;move&quot; operation is hidden.</p>

<p>.NET's IEnumerator is a closer analog to Clojure sequences than Java's Iterator, because it separates the &quot;get current&quot; operation from the &quot;move&quot; operation.  So you can call &quot;get current&quot; multiple times with the same result, like Clojure's &quot;first&quot;.</p>

<p>Can we create a purely-functional observer?  An observer could return a modified version of itself, which the Observable would use to process the next event.  This implies that events must be sequentialized, but that's what people expect anyway.  It also implies that the observable needs to keep track of that return value.  This limits the stateful parts of the model to sources of events (and possibly the final sinks to which they go).  Just because sources and sinks are mutable doesn't mean that everything on the chain between them has to be mutable.</p>

<p>Haskell's Iteratee is an attempt to solve the same resource-management problems of lazy seqs, by putting the collection in charge of the iteration.</p>

<p>Clojure has the IReduce interface for &quot;internal&quot; reduce operations.  Instead of realizing a sequence and reducing over it, you pass a function to a collection and tell it to reduce over itself, which is more efficient.  But it can't short-circuit; there's no way for the function to signal to the collection that it is not interested in receiving any more values.  If we had a purely-functional observer, we might be able to unify it with internal reduce.</p>

<h2>Ideas: April 23, 2011</h2>

<p>Expressed in <a href="https://github.com/stuartsierra/cljque/blob/501a6340204aae085a8c909de6cf64f1a3f1a65c/modules/cljque-base/src/main/clojure/cljque/active_observers.clj">this source file</a></p>

<p>Start by calling <code>(observe observable observer)</code>, which returns immediately. In another thread, the observable will call <code>(on-next observer event)</code> where <code>event</code> is a view of the observable supporting the method <code>current</code>, which either returns the latest value of the observable or throws an exception. An <code>event</code> of <code>nil</code> signals to the observer that the observable is finished generating events.</p>

<p>The observer's <code>on-next</code> method returns an updated observer to the observable. An observer supports a <code>more?</code> method to signal to the observable whether or not it wants to receive more events. When <code>more?</code> returns false, the observer must also support a <code>result</code> method. The <code>result</code> method returns a final value computed by this observer (such as an aggregation or reduce), or throws an exception.</p>

<p>All the methods are pure except for <code>observe</code>, which may initiate a computation in another thread. Combinators such as <code>map</code>, <code>reduce</code>, and <code>filter</code> are applied to observers instead of observables.</p>

<p>The methods can be summarized as follows:</p>

<ul>
	<li>Observable
	<ul>
		<li><code>(observe observable observer) ;;=&gt; updated observable</code></li>
		<li><code>(current observable) ;;=&gt; current value or throws exception</code></li>
	</ul>
	</li>
	<li>Observer
	<ul>
		<li><code>(on-next observer observable) ;;=&gt; updated observer</code></li>
		<li><code>(result observer) ;;=&gt; final value or throws exception</code></li>
		<li><code>(more? observer) ;;=&gt; boolean</code></li>
	</ul>
	</li>
</ul>

<p>The methods roughly correspond to lazy sequences as follows:</p>

<table class="confluenceTable"><tbody>
<tr>
<th class="confluenceTh"><p> Observable </p></th>
<th class="confluenceTh"><p> Observer </p></th>
<th class="confluenceTh"><p> Seq </p></th>
</tr>
<tr>
<td class="confluenceTd"><p> observe </p></td>
<td class="confluenceTd"><p> on-next </p></td>
<td class="confluenceTd"><p> seq </p></td>
</tr>
<tr>
<td class="confluenceTd"><p> current </p></td>
<td class="confluenceTd"><p> result </p></td>
<td class="confluenceTd"><p> first </p></td>
</tr>
<tr>
<td class="confluenceTd"><p> not nil? </p></td>
<td class="confluenceTd"><p> more? </p></td>
<td class="confluenceTd"><p> rest </p></td>
</tr>
</tbody></table>

<p>Where I'm currently stuck is on the return value of <code>observe</code>.  Logically, it should return another Observable, containing some reference to the <code>result</code> of the passed observer.  I think it should return an &quot;Observable Future,&quot; which can be dereferenced to get the <code>result</code> value of the observer, or observed again to subscribe another observer to that result when it becomes available.</p>

<h2>Ideas: July 9, 2011: &quot;Future Seqs&quot;</h2>

<ul>
	<li>Add callbacks to promises: invoke a fn when the promise gets a value</li>
	<li>Construct a &quot;future seq&quot; where each Cons cell is backed by a promise</li>
	<li><a href="https://github.com/stuartsierra/cljque/blob/feb58d1845569678ca4049e05c70699a5821d62b/src/main/clojure/cljque/inotify.clj">Example implementation here</a></li>
</ul>