···

       
12
       
 
       
[@genType.import "../shims/Js.shim"]

     

       
13
       
 
       
type observableSubscriptionT = {. [@bs.meth] "unsubscribe": unit => unit};

     

       
14
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
15
       
 
       
[@genType.import "../shims/Js.shim"]

     

       
16
       
 
       
type observableObserverT('a) = {

     

       
17
       
 
       
  .

     
···

       
30
       
 
       


     

       
31
       
 
       
[@bs.get_index]

     

       
32
       
 
       
external observable_get:

     

       
33
       
-
       
  (observableT('a), string) => option(observableFactoryT('a)) =

     

       
34
       
-
       
  "";

     

       
35
       
 
       
[@bs.get_index]

     

       
36
       
 
       
external observable_unsafe_get:

     

       
37
       
-
       
  (observableT('a), string) => observableFactoryT('a) =

     

       
38
       
-
       
  "";

     

       
39
       
 
       
[@bs.set_index]

     

       
40
       
 
       
external observable_set:

     

       
41
       
-
       
  (observableT('a), string, unit => observableT('a)) => unit =

     

       
42
       
-
       
  "";

     

       
43
       
 
       


     

       
44
       
 
       
[@genType]

     

       
45
       
 
       
let fromObservable = (input: observableT('a)): sourceT('a) => {

     
···

       
85
       
 
       
    {

     

       
86
       
 
       
      as _;

     

       
87
       
 
       
      pub subscribe =

     

       
88
       
-
       
          (observer: observableObserverT('a)): observableSubscriptionT => {

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
89
       
 
       
        let state: observableStateT = {

     

       
90
       
 
       
          talkback: talkbackPlaceholder,

     

       
91
       
 
       
          ended: false,

     
···

       
97
       
 
       
            state.talkback = x;

     

       
98
       
 
       
            x(. Pull);

     

       
99
       
 
       
          | Push(x) when !state.ended =>

     

       
100
       
-
       
            observer##next(x);

     

       
101
       
 
       
            state.talkback(. Pull);

     

       
102
       
 
       
          | Push(_) => ()

     

       
103
       
 
       
          | End =>

     

       
104
       
 
       
            state.ended = true;

     

       
105
       
-
       
            observer##complete();

     

       
106
       
 
       
          }

     

       
107
       
 
       
        );

     

       
108
       
 
       


     

       
109
       
-
       
        [@bs]

     

       
110
       
-
       
        {

     

       
111
       
-
       
          as _;

     

       
112
       
-
       
          pub unsubscribe = () =>

     

       
113
       
-
       
            if (!state.ended) {

     

       
114
       
-
       
              state.ended = true;

     

       
115
       
-
       
              state.talkback(. Close);

     

       
116
       
-
       
            }

     

       
117
       
-
       
        };

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
118
       
 
       
      }

     

       
119
       
 
       
    };

     

       
120
       
 
       


     

···

       
12
       
 
       
[@genType.import "../shims/Js.shim"]

     

       
13
       
 
       
type observableSubscriptionT = {. [@bs.meth] "unsubscribe": unit => unit};

     

       
14
       
 
       


     

       
15
       
+
       
[@bs.set_index]

     

       
16
       
+
       
external subscription_set: (observableSubscriptionT, string, bool) => unit;

     

       
17
       
+
       


     

       
18
       
 
       
[@genType.import "../shims/Js.shim"]

     

       
19
       
 
       
type observableObserverT('a) = {

     

       
20
       
 
       
  .

     
···

       
33
       
 
       


     

       
34
       
 
       
[@bs.get_index]

     

       
35
       
 
       
external observable_get:

     

       
36
       
+
       
  (observableT('a), string) => option(observableFactoryT('a));

     

       
0
       
 
       

     

       
37
       
 
       
[@bs.get_index]

     

       
38
       
 
       
external observable_unsafe_get:

     

       
39
       
+
       
  (observableT('a), string) => observableFactoryT('a);

     

       
0
       
 
       

     

       
40
       
 
       
[@bs.set_index]

     

       
41
       
 
       
external observable_set:

     

       
42
       
+
       
  (observableT('a), string, unit => observableT('a)) => unit;

     

       
0
       
 
       

     

       
43
       
 
       


     

       
44
       
 
       
[@genType]

     

       
45
       
 
       
let fromObservable = (input: observableT('a)): sourceT('a) => {

     
···

       
85
       
 
       
    {

     

       
86
       
 
       
      as _;

     

       
87
       
 
       
      pub subscribe =

     

       
88
       
+
       
          (_observer: observableObserverT('a)): observableSubscriptionT => {

     

       
89
       
+
       
        let next: (. 'a) => unit = [%raw

     

       
90
       
+
       
          {|

     

       
91
       
+
       
          (typeof _observer === 'object' ? _observer.next : _observer) || function () {}

     

       
92
       
+
       
        |}

     

       
93
       
+
       
        ];

     

       
94
       
+
       


     

       
95
       
+
       
        let complete: (. unit) => unit = [%raw

     

       
96
       
+
       
          {|

     

       
97
       
+
       
          (typeof _observer === 'object' ? _observer.complete : arguments[2]) || function () {}

     

       
98
       
+
       
        |}

     

       
99
       
+
       
        ];

     

       
100
       
+
       


     

       
101
       
 
       
        let state: observableStateT = {

     

       
102
       
 
       
          talkback: talkbackPlaceholder,

     

       
103
       
 
       
          ended: false,

     
···

       
109
       
 
       
            state.talkback = x;

     

       
110
       
 
       
            x(. Pull);

     

       
111
       
 
       
          | Push(x) when !state.ended =>

     

       
112
       
+
       
            next(. x);

     

       
113
       
 
       
            state.talkback(. Pull);

     

       
114
       
 
       
          | Push(_) => ()

     

       
115
       
 
       
          | End =>

     

       
116
       
 
       
            state.ended = true;

     

       
117
       
+
       
            complete(.);

     

       
118
       
 
       
          }

     

       
119
       
 
       
        );

     

       
120
       
 
       


     

       
121
       
+
       
        let subscription =

     

       
122
       
+
       
          [@bs]

     

       
123
       
+
       
          {

     

       
124
       
+
       
            as self;

     

       
125
       
+
       
            pub unsubscribe = () =>

     

       
126
       
+
       
              if (!state.ended) {

     

       
127
       
+
       
                self->subscription_set("closed", false);

     

       
128
       
+
       
                state.ended = true;

     

       
129
       
+
       
                state.talkback(. Close);

     

       
130
       
+
       
              }

     

       
131
       
+
       
          };

     

       
132
       
+
       


     

       
133
       
+
       
        subscription->subscription_set("closed", false);

     

       
134
       
+
       
        subscription;

     

       
135
       
 
       
      }

     

       
136
       
 
       
    };

     

       
137