Is there any pattern I can use to handle Event Streams with Fluture?

[krzysztofpniak]

I know there are libraries like @most/core, but is seems they don't implement fantasy-land.
I'm wondering if it is possible to handle streams with fluture. Is there any pattern I can use?

[Avaq]

You can use a lazy sequence of Futures to approximate a Stream.

You could use, for example, something like https://github.com/francisrstokes/Lazy-Infinite-List, and have a producer that emits Futures, process the stream by mapping, and consuming with go. Like so:

const Infinite = require ('lazy-infinite')
const fl = require ('fluture')

const stream = Infinite.of (function*() {
  let x = 0
  while (true) yield Future.after (300) (x++)
})

// to process items in the stream, we can just map-map:
const processedStream = fl.map (fl.map (x => `This is number ${x}`)) (stream)

// we can also map-chain to process with a side effect or async thing:
const effectStream = fl.map (fl.chain (fl.encase (x => {
  console.log (x)
}))) (processedStream)

// and then to consume, we can use go
fl.promise (fl.go (effectStream.gen))

---

In theory you should also be able to get Stream behaviour by using a List Monad Transformer:

const Stream = List.Transformer (Future)

But I don't know of any packages that implement the list transformer. 🤷‍♂️

[Avaq]

See this comment #478 (reply in thread) below for an expanded example of using this approach to model a stream of DOM events.

[krzysztofpniak]

@Avaq Thanks for the quick response. Your advice looks very interesting. I'm not closing the issue for now because it is very fresh. Let's see if there are any other approaches.

[krzysztofpniak]

@Avaq I realized that your solution produces a constant stream of data. Could you guide me how to implement a source of real events like button clicks?

[krzysztofpniak]

Anyways I found a solution.

const fromEvent = subscribe =>
  Future((reject, resolve) => {
    return subscribe(resolve);
  });

const eventStream = subscribe =>
  Infinite.of(function*() {
    let currentEmitter = () => {};
    subscribe(x => currentEmitter(x));
    const localSubscribe = emitter => {
      currentEmitter = emitter;
      return () => {};
    };
    while (true) {
      yield fromEvent(localSubscribe);
    }
  });

const stream = eventStream(emitter => {
  document.addEventListener("keyup", e => emitter(e));
  return document.removeEventListener("keyup", emitter);
});

// to process items in the stream, we can just map-map:
const processedStream = map(map(x => `Key Code = ${x.keyCode}`))(stream);

// we can also map-chain to process with a side effect or async thing:
const effectStream = map(
  chain(
    encase(x => {
      console.log(x);
    })
  )
)(processedStream);

// and then to consume, we can use go
promise(go(effectStream.gen));

Maybe it will help someone. Please note that unsubscribe is not fully implemented.

[Avaq]

I know I'm a year late, but since I dug up this issue and converted it to a Discussion, I'd like to chime in for completeness sake. I think I'd model a stream of button-clicks somewhat like this:

const Future = require ('fluture');
const Infinite = require ('lazy-infinite');

//    once :: Key k => k -> EventEmitter k v -> Future a v
//         -- Creates a Future that listens to a single event emitted on an emitter,
//         -- and then removes itself
const once = key => emitter => Future ((rej, res) => {
  emitter.addEventListener (key, handler);
  const cancel = () => emitter.removeEventListener (key, handler);
  function handler(e) {
    cancel ();
    res (e);
  }
  return cancel;
});

// The stream of keyups is just a never-ending lazy list of single emits of
// the keyup event
const keyupStream = Infinite.generator (function*() {
  while (true) yield once ('keyup') (document);
});

// A function which logs items in a Stream to the console.
const logStreamItems = Future.map (Future.chain (Future.encase (e => {
  console.log (e);
})));

// Putting our generator in `Future.go` makes it so the Futures in the
// list are consumed sequentially
const main = Future.go (Infinite.toGenerator (logStreamItems (keyupStream)))

// Our cancel function actually cancels the whole stream, thanks to
// cancellation propagation handled by Future.go and our "once" function.
const cancel = Future.fork (console.error) (console.log) (main)

// For our demo, let's log keyup events for 20 seconds, and then stop.
setTimeout (cancel, 20000);

[Avaq]

You can see it in action here: https://scrimba.com/scrim/cWWGz7fb