From 2d4add3fb769baf17a34c6a7e36dde68102a5a22 Mon Sep 17 00:00:00 2001
From: Stefano Pigozzi <stefano.pigozzi@gmail.com>
Date: Sat, 30 Dec 2017 11:49:48 +0100
Subject: mruby: add a basic ruby implementation of obsevables

---
 player/mruby/observable.mrb | 73 +++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 73 insertions(+)
 create mode 100644 player/mruby/observable.mrb

(limited to 'player/mruby/observable.mrb')

diff --git a/player/mruby/observable.mrb b/player/mruby/observable.mrb
new file mode 100644
index 0000000000..3b97461dd0
--- /dev/null
+++ b/player/mruby/observable.mrb
@@ -0,0 +1,73 @@
+# inspired by toy-rx: https://github.com/staltz/toy-rx
+class Subscriber
+  attr_writer :subscription
+
+  def initialize(observer)
+    @observer = observer
+  end
+
+  def next(value)
+    @observer.next(value)
+  end
+
+  def error(error)
+    @observer.error(error)
+  end
+
+  def complete
+    @observer.complete
+  end
+
+  def unsubscribe
+    @subscription.unsubscribe unless @subscription.nil?
+  end
+end
+
+class Observable
+  def initialize(&subscribe)
+    @subscribe = subscribe
+  end
+
+  def self.create(&subscribe)
+    fail ArgumentError, "`create` needs a subscribe block" unless block_given?
+    new do |observer|
+      subscriber = Subscriber.new(observer)
+      subscription = subscribe.call(subscriber);
+      subscriber.subscription = subscription
+      subscription
+    end
+  end
+
+  # XXX probably nicer to define operators in another file
+  def map(&tx)
+    inobservable = self
+    Observable.create do |out|
+      observer = {
+        next: -> (x) {
+          begin
+            out.next(tx.call(x))
+          rescue => e
+            out.error(e)
+          end
+        },
+        error: -> (e) { out.error(e) },
+        complete: -> (e) { out.complete },
+      };
+      inobservable.subscribe(observer)
+    end
+  end
+
+  def subscribe(callbacks = {}, &nextcb)
+    callbacks[:next] = nextcb if block_given?
+    noop = -> () {}
+
+    klass = Class.new do
+      define_method(:next,     callbacks.fetch(:next, noop))
+      define_method(:error,    callbacks.fetch(:error, noop))
+      define_method(:complete, callbacks.fetch(:complete, noop))
+    end
+
+    observer = klass.new
+    @subscribe.call(observer)
+  end
+end
-- 
cgit v1.2.3