diff options
Diffstat (limited to '2024/09/12/Functional-Reactive-Programming-in-F')
| -rw-r--r-- | 2024/09/12/Functional-Reactive-Programming-in-F/index.html | 37 |
1 files changed, 10 insertions, 27 deletions
diff --git a/2024/09/12/Functional-Reactive-Programming-in-F/index.html b/2024/09/12/Functional-Reactive-Programming-in-F/index.html index 94586aea..b1bc57f8 100644 --- a/2024/09/12/Functional-Reactive-Programming-in-F/index.html +++ b/2024/09/12/Functional-Reactive-Programming-in-F/index.html @@ -209,7 +209,8 @@ <p>We saw in the previous post on message queues that one of the advantages of that approach was that the requests were “serialized” making it conceptually easier to deal with.</p> <p>There is a similar approach that can be used with events. The idea is to turn a series of events into an “event stream”. Event streams then become quite like IEnumerables, and so the obvious next step is to treat them in much the the same way that LINQ handles collections, so that they can be filtered, mapped, split and combined.</p> <p>F# has built in support for this model, as well as for the more traditional approach.</p> -<h2 id="A-simple-event-stream"><a href="#A-simple-event-stream" class="headerlink" title="A simple event stream"></a>A simple event stream</h2><p>Let’s start with a simple example to compare the two approaches. We’ll implement the classic event handler approach first.</p> +<h2 id="A-simple-event-stream"><a class="header-anchor" href="#A-simple-event-stream">¶</a>A simple event stream</h2> +<p>Let’s start with a simple example to compare the two approaches. We’ll implement the classic event handler approach first.</p> <p>First, we define a utility function that will:</p> <ul> <li>create a timer</li> @@ -218,60 +219,47 @@ </ul> <p>Here’s the code:</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">open</span> System</span><br><span class="line"><span class="keyword">open</span> System.Threading</span><br><span class="line"></span><br><span class="line"><span class="comment">/// create a timer and register an event handler,</span></span><br><span class="line"><span class="comment">/// then run the timer for five seconds</span></span><br><span class="line"><span class="keyword">let</span> createTimer timerInterval eventHandler <span class="operator">=</span></span><br><span class="line"> <span class="comment">// setup a timer</span></span><br><span class="line"> <span class="keyword">let</span> timer <span class="operator">=</span> <span class="keyword">new</span> System.Timers.Timer(float timerInterval)</span><br><span class="line"> timer.AutoReset <span class="operator"><-</span> <span class="literal">true</span></span><br><span class="line"></span><br><span class="line"> <span class="comment">// add an event handler</span></span><br><span class="line"> timer.Elapsed.Add eventHandler</span><br><span class="line"></span><br><span class="line"> <span class="comment">// return an async task</span></span><br><span class="line"> <span class="keyword">async</span> {</span><br><span class="line"> <span class="comment">// start timer...</span></span><br><span class="line"> timer.Start()</span><br><span class="line"> <span class="comment">// ...run for five seconds...</span></span><br><span class="line"> <span class="keyword">do!</span> Async.Sleep <span class="number">5000</span></span><br><span class="line"> <span class="comment">// ... and stop</span></span><br><span class="line"> timer.Stop()</span><br><span class="line"> }</span><br></pre></td></tr></table></figure> - <p>Now test it interactively:</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// create a handler. The event args are ignored</span></span><br><span class="line"><span class="keyword">let</span> basicHandler _ <span class="operator">=</span> <span class="built_in">printfn</span> <span class="string">"tick %A"</span> DateTime.Now</span><br><span class="line"></span><br><span class="line"><span class="comment">// register the handler</span></span><br><span class="line"><span class="keyword">let</span> basicTimer1 <span class="operator">=</span> createTimer <span class="number">1000</span> basicHandler</span><br><span class="line"></span><br><span class="line"><span class="comment">// run the task now</span></span><br><span class="line">Async.RunSynchronously basicTimer1</span><br></pre></td></tr></table></figure> - <p>Now let’s create a similar utility method to create a timer, but this time it will return an “observable” as well, which is the stream of events.</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">let</span> createTimerAndObservable timerInterval <span class="operator">=</span></span><br><span class="line"> <span class="comment">// setup a timer</span></span><br><span class="line"> <span class="keyword">let</span> timer <span class="operator">=</span> <span class="keyword">new</span> System.Timers.Timer(float timerInterval)</span><br><span class="line"> timer.AutoReset <span class="operator"><-</span> <span class="literal">true</span></span><br><span class="line"></span><br><span class="line"> <span class="comment">// events are automatically IObservable</span></span><br><span class="line"> <span class="keyword">let</span> observable <span class="operator">=</span> timer.Elapsed</span><br><span class="line"></span><br><span class="line"> <span class="comment">// return an async task</span></span><br><span class="line"> <span class="keyword">let</span> task <span class="operator">=</span> <span class="keyword">async</span> {</span><br><span class="line"> timer.Start()</span><br><span class="line"> <span class="keyword">do!</span> Async.Sleep <span class="number">5000</span></span><br><span class="line"> timer.Stop()</span><br><span class="line"> }</span><br><span class="line"></span><br><span class="line"> <span class="comment">// return a async task and the observable</span></span><br><span class="line"> (task,observable)</span><br></pre></td></tr></table></figure> - <p>And again test it interactively:</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// create the timer and the corresponding observable</span></span><br><span class="line"><span class="keyword">let</span> basicTimer2 , timerEventStream <span class="operator">=</span> createTimerAndObservable <span class="number">1000</span></span><br><span class="line"></span><br><span class="line"><span class="comment">// register that every time something happens on the</span></span><br><span class="line"><span class="comment">// event stream, print the time.</span></span><br><span class="line">timerEventStream</span><br><span class="line"><span class="operator">|></span> Observable.subscribe (<span class="keyword">fun</span> _ <span class="operator">-></span> <span class="built_in">printfn</span> <span class="string">"tick %A"</span> DateTime.Now)</span><br><span class="line"></span><br><span class="line"><span class="comment">// run the task now</span></span><br><span class="line">Async.RunSynchronously basicTimer2</span><br></pre></td></tr></table></figure> - <p>The difference is that instead of registering a handler directly with an event, we are “subscribing” to an event stream. Subtly different, and important.</p> -<h2 id="Counting-events"><a href="#Counting-events" class="headerlink" title="Counting events"></a>Counting events</h2><p>In this next example, we’ll have a slightly more complex requirement:</p> +<h2 id="Counting-events"><a class="header-anchor" href="#Counting-events">¶</a>Counting events</h2> +<p>In this next example, we’ll have a slightly more complex requirement:</p> <figure class="highlight plaintext"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line">Create a timer that ticks every 500ms.</span><br><span class="line">At each tick, print the number of ticks so far and the current time.</span><br></pre></td></tr></table></figure> - <p>To do this in a classic imperative way, we would probably create a class with a mutable counter, as below:</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">type</span> <span class="title class_">ImperativeTimerCount</span>() <span class="operator">=</span></span><br><span class="line"></span><br><span class="line"> <span class="keyword">let</span> <span class="keyword">mutable</span> count <span class="operator">=</span> <span class="number">0</span></span><br><span class="line"></span><br><span class="line"> <span class="comment">// the event handler. The event args are ignored</span></span><br><span class="line"> <span class="keyword">member</span> this.handleEvent _ <span class="operator">=</span></span><br><span class="line"> count <span class="operator"><-</span> count <span class="operator">+</span> <span class="number">1</span></span><br><span class="line"> <span class="built_in">printfn</span> <span class="string">"timer ticked with count %i"</span> count</span><br></pre></td></tr></table></figure> - <p>We can reuse the utility functions we created earlier to test it:</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// create a handler class</span></span><br><span class="line"><span class="keyword">let</span> handler <span class="operator">=</span> <span class="keyword">new</span> ImperativeTimerCount()</span><br><span class="line"></span><br><span class="line"><span class="comment">// register the handler method</span></span><br><span class="line"><span class="keyword">let</span> timerCount1 <span class="operator">=</span> createTimer <span class="number">500</span> handler.handleEvent</span><br><span class="line"></span><br><span class="line"><span class="comment">// run the task now</span></span><br><span class="line">Async.RunSynchronously timerCount1</span><br></pre></td></tr></table></figure> - <p>Let’s see how we would do this same thing in a functional way:</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// create the timer and the corresponding observable</span></span><br><span class="line"><span class="keyword">let</span> timerCount2, timerEventStream <span class="operator">=</span> createTimerAndObservable <span class="number">500</span></span><br><span class="line"></span><br><span class="line"><span class="comment">// set up the transformations on the event stream</span></span><br><span class="line">timerEventStream</span><br><span class="line"><span class="operator">|></span> Observable.scan (<span class="keyword">fun</span> count _ <span class="operator">-></span> count <span class="operator">+</span> <span class="number">1</span>) <span class="number">0</span></span><br><span class="line"><span class="operator">|></span> Observable.subscribe (<span class="keyword">fun</span> count <span class="operator">-></span> <span class="built_in">printfn</span> <span class="string">"timer ticked with count %i"</span> count)</span><br><span class="line"></span><br><span class="line"><span class="comment">// run the task now</span></span><br><span class="line">Async.RunSynchronously timerCount2</span><br></pre></td></tr></table></figure> - <p>Here we see how you can build up layers of event transformations, just as you do with list transformations in LINQ.</p> <p>The first transformation is <code>scan</code>, which accumulates state for each event. It is roughly equivalent to the <code>List.fold</code> function that we have seen used with lists. In this case, the accumulated state is just a counter.</p> <p>And then, for each event, the count is printed out.</p> <p>Note that in this functional approach, we didn’t have any mutable state, and we didn’t need to create any special classes.</p> -<h2 id="Merging-multiple-event-streams"><a href="#Merging-multiple-event-streams" class="headerlink" title="Merging multiple event streams"></a>Merging multiple event streams</h2><p>For a final example, we’ll look at merging multiple event streams.</p> +<h2 id="Merging-multiple-event-streams"><a class="header-anchor" href="#Merging-multiple-event-streams">¶</a>Merging multiple event streams</h2> +<p>For a final example, we’ll look at merging multiple event streams.</p> <p>Let’s make a requirement based on the well-known “FizzBuzz” problem:</p> <figure class="highlight plaintext"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br></pre></td><td class="code"><pre><span class="line">Create two timers, called '3' and '5'. The '3' timer ticks every 300ms and the '5' timer ticks</span><br><span class="line">every 500ms.</span><br><span class="line"></span><br><span class="line">Handle the events as follows:</span><br><span class="line">a) for all events, print the id of the time and the time</span><br><span class="line">b) when a tick is simultaneous with a previous tick, print 'FizzBuzz'</span><br><span class="line">otherwise:</span><br><span class="line">c) when the '3' timer ticks on its own, print 'Fizz'</span><br><span class="line">d) when the '5' timer ticks on its own, print 'Buzz'</span><br></pre></td></tr></table></figure> - <p>First let’s create some code that both implementations can use.</p> <p>We’ll want a generic event type that captures the timer id and the time of the tick.</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">type</span> <span class="title class_">FizzBuzzEvent</span> <span class="operator">=</span> {label<span class="operator">:</span><span class="type">int</span>; time<span class="operator">:</span> DateTime}</span><br></pre></td></tr></table></figure> - <p>And then we need a utility function to see if two events are simultaneous. We’ll be generous and allow a time difference of up to 50ms.</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">let</span> areSimultaneous (earlierEvent,laterEvent) <span class="operator">=</span></span><br><span class="line"> <span class="keyword">let</span> {label<span class="operator">=</span>_;time<span class="operator">=</span>t1} <span class="operator">=</span> earlierEvent</span><br><span class="line"> <span class="keyword">let</span> {label<span class="operator">=</span>_;time<span class="operator">=</span>t2} <span class="operator">=</span> laterEvent</span><br><span class="line"> t2.Subtract(t1).Milliseconds <span class="operator"><</span> <span class="number">50</span></span><br></pre></td></tr></table></figure> - <p>In the imperative design, we’ll need to keep track of the previous event, so we can compare them. And we’ll need special case code for the first time, when the previous event doesn’t exist</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">type</span> <span class="title class_">ImperativeFizzBuzzHandler</span>() <span class="operator">=</span></span><br><span class="line"></span><br><span class="line"> <span class="keyword">let</span> <span class="keyword">mutable</span> previousEvent<span class="operator">:</span> FizzBuzzEvent <span class="type">option</span> <span class="operator">=</span> <span class="literal">None</span></span><br><span class="line"></span><br><span class="line"> <span class="keyword">let</span> printEvent thisEvent <span class="operator">=</span></span><br><span class="line"> <span class="keyword">let</span> {label<span class="operator">=</span><span class="built_in">id</span>; time<span class="operator">=</span>t} <span class="operator">=</span> thisEvent</span><br><span class="line"> <span class="built_in">printf</span> <span class="string">"[%i] %i.%03i "</span> <span class="built_in">id</span> t.Second t.Millisecond</span><br><span class="line"> <span class="keyword">let</span> simultaneous <span class="operator">=</span> previousEvent.IsSome <span class="operator">&&</span> areSimultaneous (previousEvent.Value,thisEvent)</span><br><span class="line"> <span class="keyword">if</span> simultaneous <span class="keyword">then</span> <span class="built_in">printfn</span> <span class="string">"FizzBuzz"</span></span><br><span class="line"> <span class="keyword">elif</span> <span class="built_in">id</span> <span class="operator">=</span> <span class="number">3</span> <span class="keyword">then</span> <span class="built_in">printfn</span> <span class="string">"Fizz"</span></span><br><span class="line"> <span class="keyword">elif</span> <span class="built_in">id</span> <span class="operator">=</span> <span class="number">5</span> <span class="keyword">then</span> <span class="built_in">printfn</span> <span class="string">"Buzz"</span></span><br><span class="line"></span><br><span class="line"> <span class="keyword">member</span> this.handleEvent3 eventArgs <span class="operator">=</span></span><br><span class="line"> <span class="keyword">let</span> event <span class="operator">=</span> {label<span class="operator">=</span><span class="number">3</span>; time<span class="operator">=</span>DateTime.Now}</span><br><span class="line"> printEvent event</span><br><span class="line"> previousEvent <span class="operator"><-</span> <span class="literal">Some</span> event</span><br><span class="line"></span><br><span class="line"> <span class="keyword">member</span> this.handleEvent5 eventArgs <span class="operator">=</span></span><br><span class="line"> <span class="keyword">let</span> event <span class="operator">=</span> {label<span class="operator">=</span><span class="number">5</span>; time<span class="operator">=</span>DateTime.Now}</span><br><span class="line"> printEvent event</span><br><span class="line"> previousEvent <span class="operator"><-</span> <span class="literal">Some</span> event</span><br></pre></td></tr></table></figure> - <p>Now the code is beginning to get ugly fast! Already we have mutable state, complex conditional logic, and special cases, just for such a simple requirement.</p> <p>Let’s test it:</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// create the class</span></span><br><span class="line"><span class="keyword">let</span> handler <span class="operator">=</span> <span class="keyword">new</span> ImperativeFizzBuzzHandler()</span><br><span class="line"></span><br><span class="line"><span class="comment">// create the two timers and register the two handlers</span></span><br><span class="line"><span class="keyword">let</span> timer3 <span class="operator">=</span> createTimer <span class="number">300</span> handler.handleEvent3</span><br><span class="line"><span class="keyword">let</span> timer5 <span class="operator">=</span> createTimer <span class="number">500</span> handler.handleEvent5</span><br><span class="line"></span><br><span class="line"><span class="comment">// run the two timers at the same time</span></span><br><span class="line">[timer3;timer5]</span><br><span class="line"><span class="operator">|></span> Async.Parallel</span><br><span class="line"><span class="operator">|></span> Async.RunSynchronously</span><br></pre></td></tr></table></figure> - <p>It does work, but are you sure the code is not buggy? Are you likely to accidentally break something if you change it?</p> <p>The problem with this imperative code is that it has a lot of noise that obscures the the requirements.</p> <p>Can the functional version do better? Let’s see!</p> <p>First, we create <em>two</em> event streams, one for each timer:</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">let</span> timer3, timerEventStream3 <span class="operator">=</span> createTimerAndObservable <span class="number">300</span></span><br><span class="line"><span class="keyword">let</span> timer5, timerEventStream5 <span class="operator">=</span> createTimerAndObservable <span class="number">500</span></span><br></pre></td></tr></table></figure> - <p>Next, we convert each event on the “raw” event streams into our FizzBuzz event type:</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// convert the time events into FizzBuzz events with the appropriate id</span></span><br><span class="line"><span class="keyword">let</span> eventStream3 <span class="operator">=</span></span><br><span class="line"> timerEventStream3</span><br><span class="line"> <span class="operator">|></span> Observable.map (<span class="keyword">fun</span> _ <span class="operator">-></span> {label<span class="operator">=</span><span class="number">3</span>; time<span class="operator">=</span>DateTime.Now})</span><br><span class="line"></span><br><span class="line"><span class="keyword">let</span> eventStream5 <span class="operator">=</span></span><br><span class="line"> timerEventStream5</span><br><span class="line"> <span class="operator">|></span> Observable.map (<span class="keyword">fun</span> _ <span class="operator">-></span> {label<span class="operator">=</span><span class="number">5</span>; time<span class="operator">=</span>DateTime.Now})</span><br></pre></td></tr></table></figure> - <p>Now, to see if two events are simultaneous, we need to compare them from the two different streams somehow.</p> <p>It’s actually easier than it sounds, because we can:</p> <ul> @@ -282,26 +270,21 @@ </ul> <p>Here’s the actual code to do this:</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// combine the two streams</span></span><br><span class="line"><span class="keyword">let</span> combinedStream <span class="operator">=</span></span><br><span class="line"> Observable.merge eventStream3 eventStream5</span><br><span class="line"></span><br><span class="line"><span class="comment">// make pairs of events</span></span><br><span class="line"><span class="keyword">let</span> pairwiseStream <span class="operator">=</span></span><br><span class="line"> combinedStream <span class="operator">|></span> Observable.pairwise</span><br><span class="line"></span><br><span class="line"><span class="comment">// split the stream based on whether the pairs are simultaneous</span></span><br><span class="line"><span class="keyword">let</span> simultaneousStream, nonSimultaneousStream <span class="operator">=</span></span><br><span class="line"> pairwiseStream <span class="operator">|></span> Observable.partition areSimultaneous</span><br></pre></td></tr></table></figure> - <p>Finally, we can split the <code>nonSimultaneousStream</code> again, based on the event id:</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// split the non-simultaneous stream based on the id</span></span><br><span class="line"><span class="keyword">let</span> fizzStream, buzzStream <span class="operator">=</span></span><br><span class="line"> nonSimultaneousStream</span><br><span class="line"> <span class="comment">// convert pair of events to the first event</span></span><br><span class="line"> <span class="operator">|></span> Observable.map (<span class="keyword">fun</span> (ev1,_) <span class="operator">-></span> ev1)</span><br><span class="line"> <span class="comment">// split on whether the event id is three</span></span><br><span class="line"> <span class="operator">|></span> Observable.partition (<span class="keyword">fun</span> {label<span class="operator">=</span><span class="built_in">id</span>} <span class="operator">-></span> <span class="built_in">id</span><span class="operator">=</span><span class="number">3</span>)</span><br></pre></td></tr></table></figure> - <p>Let’s review so far. We have started with the two original event streams and from them created four new ones:</p> <ul> <li><code>combinedStream</code> contains all the events</li> <li><code>simultaneousStream</code> contains only the simultaneous events</li> -<li><code>fizzStream</code> contains only the non-simultaneous events with id=3</li> -<li><code>buzzStream</code> contains only the non-simultaneous events with id=5</li> +<li><code>fizzStream</code> contains only the non-simultaneous events with id=3</li> +<li><code>buzzStream</code> contains only the non-simultaneous events with id=5</li> </ul> <p>Now all we need to do is attach behavior to each stream:</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">//print events from the combinedStream</span></span><br><span class="line">combinedStream</span><br><span class="line"><span class="operator">|></span> Observable.subscribe (<span class="keyword">fun</span> {label<span class="operator">=</span><span class="built_in">id</span>;time<span class="operator">=</span>t} <span class="operator">-></span></span><br><span class="line"> <span class="built_in">printf</span> <span class="string">"[%i] %i.%03i "</span> <span class="built_in">id</span> t.Second t.Millisecond)</span><br><span class="line"></span><br><span class="line"><span class="comment">//print events from the simultaneous stream</span></span><br><span class="line">simultaneousStream</span><br><span class="line"><span class="operator">|></span> Observable.subscribe (<span class="keyword">fun</span> _ <span class="operator">-></span> <span class="built_in">printfn</span> <span class="string">"FizzBuzz"</span>)</span><br><span class="line"></span><br><span class="line"><span class="comment">//print events from the nonSimultaneous streams</span></span><br><span class="line">fizzStream</span><br><span class="line"><span class="operator">|></span> Observable.subscribe (<span class="keyword">fun</span> _ <span class="operator">-></span> <span class="built_in">printfn</span> <span class="string">"Fizz"</span>)</span><br><span class="line"></span><br><span class="line">buzzStream</span><br><span class="line"><span class="operator">|></span> Observable.subscribe (<span class="keyword">fun</span> _ <span class="operator">-></span> <span class="built_in">printfn</span> <span class="string">"Buzz"</span>)</span><br></pre></td></tr></table></figure> - <p>Let’s test it:</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// run the two timers at the same time</span></span><br><span class="line">[timer3;timer5]</span><br><span class="line"><span class="operator">|></span> Async.Parallel</span><br><span class="line"><span class="operator">|></span> Async.RunSynchronously</span><br></pre></td></tr></table></figure> - <p>Here’s all the code in one complete set:</p> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// create the event streams and raw observables</span></span><br><span class="line"><span class="keyword">let</span> timer3, timerEventStream3 <span class="operator">=</span> createTimerAndObservable <span class="number">300</span></span><br><span class="line"><span class="keyword">let</span> timer5, timerEventStream5 <span class="operator">=</span> createTimerAndObservable <span class="number">500</span></span><br><span class="line"></span><br><span class="line"><span class="comment">// convert the time events into FizzBuzz events with the appropriate id</span></span><br><span class="line"><span class="keyword">let</span> eventStream3 <span class="operator">=</span> timerEventStream3</span><br><span class="line"> <span class="operator">|></span> Observable.map (<span class="keyword">fun</span> _ <span class="operator">-></span> {label<span class="operator">=</span><span class="number">3</span>; time<span class="operator">=</span>DateTime.Now})</span><br><span class="line"><span class="keyword">let</span> eventStream5 <span class="operator">=</span> timerEventStream5</span><br><span class="line"> <span class="operator">|></span> Observable.map (<span class="keyword">fun</span> _ <span class="operator">-></span> {label<span class="operator">=</span><span class="number">5</span>; time<span class="operator">=</span>DateTime.Now})</span><br><span class="line"></span><br><span class="line"><span class="comment">// combine the two streams</span></span><br><span class="line"><span class="keyword">let</span> combinedStream <span class="operator">=</span></span><br><span class="line"> Observable.merge eventStream3 eventStream5</span><br><span class="line"></span><br><span class="line"><span class="comment">// make pairs of events</span></span><br><span class="line"><span class="keyword">let</span> pairwiseStream <span class="operator">=</span></span><br><span class="line"> combinedStream <span class="operator">|></span> Observable.pairwise</span><br><span class="line"></span><br><span class="line"><span class="comment">// split the stream based on whether the pairs are simultaneous</span></span><br><span class="line"><span class="keyword">let</span> simultaneousStream, nonSimultaneousStream <span class="operator">=</span></span><br><span class="line"> pairwiseStream <span class="operator">|></span> Observable.partition areSimultaneous</span><br><span class="line"></span><br><span class="line"><span class="comment">// split the non-simultaneous stream based on the id</span></span><br><span class="line"><span class="keyword">let</span> fizzStream, buzzStream <span class="operator">=</span></span><br><span class="line"> nonSimultaneousStream</span><br><span class="line"> <span class="comment">// convert pair of events to the first event</span></span><br><span class="line"> <span class="operator">|></span> Observable.map (<span class="keyword">fun</span> (ev1,_) <span class="operator">-></span> ev1)</span><br><span class="line"> <span class="comment">// split on whether the event id is three</span></span><br><span class="line"> <span class="operator">|></span> Observable.partition (<span class="keyword">fun</span> {label<span class="operator">=</span><span class="built_in">id</span>} <span class="operator">-></span> <span class="built_in">id</span><span class="operator">=</span><span class="number">3</span>)</span><br><span class="line"></span><br><span class="line"><span class="comment">//print events from the combinedStream</span></span><br><span class="line">combinedStream</span><br><span class="line"><span class="operator">|></span> Observable.subscribe (<span class="keyword">fun</span> {label<span class="operator">=</span><span class="built_in">id</span>;time<span class="operator">=</span>t} <span class="operator">-></span></span><br><span class="line"> <span class="built_in">printf</span> <span class="string">"[%i] %i.%03i "</span> <span class="built_in">id</span> t.Second t.Millisecond)</span><br><span class="line"></span><br><span class="line"><span class="comment">//print events from the simultaneous stream</span></span><br><span class="line">simultaneousStream</span><br><span class="line"><span class="operator">|></span> Observable.subscribe (<span class="keyword">fun</span> _ <span class="operator">-></span> <span class="built_in">printfn</span> <span class="string">"FizzBuzz"</span>)</span><br><span class="line"></span><br><span class="line"><span class="comment">//print events from the nonSimultaneous streams</span></span><br><span class="line">fizzStream</span><br><span class="line"><span class="operator">|></span> Observable.subscribe (<span class="keyword">fun</span> _ <span class="operator">-></span> <span class="built_in">printfn</span> <span class="string">"Fizz"</span>)</span><br><span class="line"></span><br><span class="line">buzzStream</span><br><span class="line"><span class="operator">|></span> Observable.subscribe (<span class="keyword">fun</span> _ <span class="operator">-></span> <span class="built_in">printfn</span> <span class="string">"Buzz"</span>)</span><br><span class="line"></span><br><span class="line"><span class="comment">// run the two timers at the same time</span></span><br><span class="line">[timer3;timer5]</span><br><span class="line"><span class="operator">|></span> Async.Parallel</span><br><span class="line"><span class="operator">|></span> Async.RunSynchronously</span><br></pre></td></tr></table></figure> - <p>The code might seem a bit long winded, but this kind of incremental, step-wise approach is very clear and self-documenting.</p> <p>Some of the benefits of this style are:</p> <ul> @@ -311,9 +294,9 @@ <li>It is easy to debug. For example, I could easily “tap” the output of the <code>simultaneousStream</code> to see if it contains what I think it contains:</li> </ul> <figure class="highlight fsharp"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// debugging code</span></span><br><span class="line"><span class="comment">//simultaneousStream |> Observable.subscribe (fun e -> printfn "sim %A" e)</span></span><br><span class="line"><span class="comment">//nonSimultaneousStream |> Observable.subscribe (fun e -> printfn "non-sim %A" e)</span></span><br></pre></td></tr></table></figure> - <p>This would be much harder in the imperative version.</p> -<h2 id="Summary"><a href="#Summary" class="headerlink" title="Summary"></a>Summary</h2><p>Functional Reactive Programming (known as FRP) is a big topic, and we’ve only just touched on it here. I hope this introduction has given you a glimpse of the usefulness of this way of doing things.</p> +<h2 id="Summary"><a class="header-anchor" href="#Summary">¶</a>Summary</h2> +<p>Functional Reactive Programming (known as FRP) is a big topic, and we’ve only just touched on it here. I hope this introduction has given you a glimpse of the usefulness of this way of doing things.</p> <p>If you want to learn more, see the documentation for the F# <a target="_blank" rel="noopener" href="https://docs.microsoft.com/en-us/previous-versions/visualstudio/visual-studio-2010/ee370313%28v=vs.100%29?redirectedfrom=MSDN">Observable module</a>, which has the basic transformations used above. And there is also the <a target="_blank" rel="noopener" href="https://docs.microsoft.com/en-us/previous-versions/dotnet/reactive-extensions/hh242985%28v=vs.103%29">Reactive Extensions (Rx)</a> library which shipped as part of .NET 4. That contains many other additional transformations.</p> </div> |