ReasonML Observables Implementation

Rex.re

type observer('a, 'e) = {
  .
  next: 'a => unit,
  complete: unit => unit,
  error: 'e => unit,
};

type subscription = {
  .
  unsubscribe: unit => unit
};

type observable('a, 'e, 'c) = {
  ..
  subscribe: observer('a, 'e) => subscription,
} as 'c;

type subject('a, 'e) = {
  .
  subscribe: observer('a, 'e) => subscription,
  next: 'a => unit,
  complete: unit => unit,
  error: 'e => unit
};

type createFn('a, 'e) = observer('a, 'e) => option(unit => unit);

let createSubscription: (unit => unit) => subscription = (onUnsubscribe) => {
  let subscription = {
    pub unsubscribe = () => onUnsubscribe();
  };

  subscription;
};

module Observer = {
  let noop = () => ();

  let create = (onNext, ~onError=noop, ~onComplete=noop, ()) => {
    let observer: observer('a, 'e) = {
      pub next = onNext;
      pub complete = onComplete;
      pub error = onError;
    };

    observer;
  };
};

module Observable = {
  let create: (createFn('a, 'e)) => observable('a, 'e, 'c) = (createFn) => {
    let observable: observable('a, 'e, 'c) = {
      pub subscribe = (observer) => {
        let subscriptionRef: ref(option(subscription)) = ref(None);
  
        let callObserver = (fn) => {
          switch (subscriptionRef^) {
            | Some(subscription) => fn(observer)
            | None => ()
            };
        };
  
        let proxyObserver = {
          pub next = value => {
            callObserver((obs) => obs#next(value));
          };
          pub complete = () => {
            callObserver((obs) => obs#complete());
          };
          pub error = (err) => {
            callObserver((obs) => obs#error(err));
          }
        };
  
        let maybeUnsubscribeFn = createFn(proxyObserver);
  
        let subscription = createSubscription(() => {
          subscriptionRef := None;
  
          switch (maybeUnsubscribeFn) {
          | Some(unsubscribeFn) => unsubscribeFn();
          | None => ()
          };
        });
  
        subscriptionRef := Some(subscription);
        subscription;
      };
    };
  
    observable;
  };
};

module Subject {
  let create: unit => subject('a, 'e) = () => {
    let subject: subject('a, 'e) = {
      val subscribers: ref(list(observer('a, 'e))) = ref([]);

      pub subscribe = (observer) => {
        let innerObservable = Observable.create(innerObserver => {
          this#addSubscriber(innerObserver);

          Some(() => {
            this#removeSubscriber(innerObserver);
          });
        });

        innerObservable#subscribe(observer);
      };

      pub next = (value: 'a) => {
        this#eachSubscriber(subscriber => subscriber#next(value));
      };

      pub complete = () => {
        this#eachSubscriber(subscriber => subscriber#complete());
      };

      pub error = (err: 'e) => {
        this#eachSubscriber(subscriber => subscriber#error(err));
      };

      pri eachSubscriber = (fn) => {
        List.iter(fn, subscribers^);
      };

      pri addSubscriber = (observer) => {
        subscribers := subscribers^ @ [observer];
      };

      pri removeSubscriber = (observer) => {
        subscribers := List.filter(subscriber => 
        subscriber !== observer, subscribers^);
      };
    };

    subject;
  };
};