feat(rx-utils): add sink2, update context2

Author: raveclassic

packages/rx-utils/src/context2.utils.ts

@@ -3,7 +3,6 @@ import { Monad2 } from 'fp-ts/lib/Monad';
 import { map as readerMap, Reader } from 'fp-ts/lib/Reader';
 import { sequenceT } from 'fp-ts/lib/Apply';
 import { Omit } from 'typelevel-ts';
-import { combineLatest, Observable } from 'rxjs';
 import {
 	ProductLeft,
 	ProductMap,
@@ -12,11 +11,10 @@ import {
 import { deferReader } from '@devexperts/utils/dist/adt/reader.utils';
 import { pipe, pipeable } from 'fp-ts/lib/pipeable';
 import { memoOnce } from '@devexperts/utils/dist/function/memoize';
-import { eqShallow } from '@devexperts/utils/dist/typeclasses/eq/eq.utils';
 import { NonEmptyArray } from 'fp-ts/lib/NonEmptyArray';
 import { array } from 'fp-ts/lib/Array';
-import { URI as URIObservable } from 'fp-ts-rxjs/lib/Observable';
-import { instanceObservable } from './observable.utils';
+import { instanceSink, sink, Sink, URI as URISink } from './sink2.utils';
+import { strictEqual } from 'fp-ts/lib/Eq';
 
 export const URI = '@devexperts/dx-utils//Context';
 export type URI = typeof URI;
@@ -26,14 +24,15 @@ declare module 'fp-ts/lib/HKT' {
 	}
 }
 
-export interface Context<E, A> extends Reader<E, Observable<A>> {}
+export interface Context<E, A> extends Reader<E, Sink<A>> {}
 
-const memo = memoOnce(eqShallow);
-export const instanceContext: Monad2<URI> & ProductLeft<URI> & ReaderM1<URIObservable> = {
+const memo = memoOnce({
+	equals: strictEqual,
+});
+export const instanceContext: Monad2<URI> & ProductLeft<URI> = {
 	URI,
-	...getReaderM(instanceObservable),
-	asks: f => memo(e => instanceObservable.of(f(e))),
-	productLeft: (fa, fb) => e => combineLatest(fa(e), fb(e)),
+	...getReaderM(instanceSink),
+	productLeft: (fa, fb) => e => sink.sequenceT(fa(e), fb(e)),
 };
 
 const sequenceT_ = sequenceT(instanceContext);
@@ -42,7 +41,7 @@ const sequenceArray = array.sequence(instanceContext);
 const defer = <E extends object, A, K extends keyof E>(
 	fa: Context<E, A>,
 	...keys: K[]
-): Context<Omit<E, K>, Context<Pick<E, K>, A>> => pipe(deferReader(fa, ...keys), readerMap(instanceObservable.of));
+): Context<Omit<E, K>, Context<Pick<E, K>, A>> => pipe(deferReader(fa, ...keys), readerMap(instanceSink.of));
 
 const combine: ProductMap<URI> = <E, A, R>(...args: NonEmptyArray<Context<E, A> | ProjectMany<A, R>>) => {
 	const last = args.length - 1;
@@ -51,11 +50,14 @@ const combine: ProductMap<URI> = <E, A, R>(...args: NonEmptyArray<Context<E, A>
 	return instanceContext.map(fas, as => project(...as));
 };
 
+const key = <A>() => <K extends PropertyKey>(key: K): Context<Record<K, A>, A> => e => sink.of(e[key]);
+
 export const context = {
 	...instanceContext,
 	...pipeable(instanceContext),
 	sequenceT: sequenceT_,
 	sequenceArray,
 	combine,
 	defer,
+	key,
 };

packages/rx-utils/src/sink.utils.ts

@@ -4,7 +4,13 @@ import { sequenceT } from 'fp-ts/lib/Apply';
 import { Monoid } from 'fp-ts/lib/Monoid';
 import { Semigroup } from 'fp-ts/lib/Semigroup';
 
+/**
+ * @deprecated Use `Sink` from `sink2.utils.ts`
+ */
 export const URI = 'Sink';
+/**
+ * @deprecated Use `Sink` from `sink2.utils.ts`
+ */
 export type URI = typeof URI;
 
 declare module 'fp-ts/lib/HKT' {
@@ -14,7 +20,7 @@ declare module 'fp-ts/lib/HKT' {
 }
 
 /**
- * @deprecated Use `MonadObservable` from `@devexperts/utils` to track effects
+ * @deprecated Use `Sink` from `sink2.utils.ts`
  */
 export class Sink<A> {
 	readonly _A!: A;
@@ -37,14 +43,14 @@ export class Sink<A> {
 }
 
 /**
- * @deprecated
+ * @deprecated use `sink.getSemigroup` from `sink2.utils.ts`
  */
 export const getSemigroup = <A>(S: Semigroup<A>): Semigroup<Sink<A>> => ({
 	concat: (x, y) => new Sink(S.concat(x.value, y.value), merge(x.sink$, y.sink$)),
 });
 
 /**
- * @deprecated
+ * @deprecated use `sink.getMonoid` from `sink2.utils.ts`
  */
 export const getMonoid = <A>(M: Monoid<A>): Monoid<Sink<A>> => ({
 	...getSemigroup(M),
@@ -57,7 +63,7 @@ const ap = <A, B>(fab: Sink<(a: A) => B>, fa: Sink<A>): Sink<B> => fa.ap(fab);
 const chain = <A, B>(fa: Sink<A>, f: (a: A) => Sink<B>): Sink<B> => fa.chain(f);
 
 /**
- * @deprecated
+ * @deprecated Use `instanceSink` from `sink2.utils.ts`
  */
 export const sink: Monad1<URI> = {
 	URI,
@@ -68,6 +74,6 @@ export const sink: Monad1<URI> = {
 };
 
 /**
- * @deprecated
+ * @deprecated Use `sink.sequenceT` from `sink2.utils.ts`
  */
 export const sequenceTSink = sequenceT(sink);

packages/rx-utils/src/sink2.utils.ts

@@ -0,0 +1,49 @@
+import { EMPTY, merge, Observable } from 'rxjs';
+import { Monad1 } from 'fp-ts/lib/Monad';
+import { pipeable } from 'fp-ts/lib/pipeable';
+import { sequenceT } from 'fp-ts/lib/Apply';
+import { array } from 'fp-ts/lib/Array';
+import { Semigroup } from 'fp-ts/lib/Semigroup';
+import { Monoid } from 'fp-ts/lib/Monoid';
+
+export interface Sink<A> {
+	readonly effects: Observable<unknown>;
+	readonly value: A;
+}
+export const newSink = <A>(value: A, effects: Observable<unknown>): Sink<A> => ({ value, effects });
+
+export const URI = '@devexperts/rx-utils//Sink';
+export type URI = typeof URI;
+declare module 'fp-ts/lib/HKT' {
+	interface URItoKind<A> {
+		[URI]: Sink<A>;
+	}
+}
+
+export const instanceSink: Monad1<URI> = {
+	URI,
+	map: (fa, f) => newSink(f(fa.value), fa.effects),
+	ap: (fab, fa) => newSink(fab.value(fa.value), merge(fab.effects, fa.effects)),
+	of: a => newSink(a, EMPTY),
+	chain: (fa, f) => {
+		const fb = f(fa.value);
+		return newSink(fb.value, merge(fa.effects, fb.effects));
+	},
+};
+
+const getSemigroup = <A>(S: Semigroup<A>): Semigroup<Sink<A>> => ({
+	concat: (x, y) => newSink(S.concat(x.value, y.value), merge(x.effects, y.effects)),
+});
+
+const getMonoid = <A>(M: Monoid<A>): Monoid<Sink<A>> => ({
+	...getSemigroup(M),
+	empty: instanceSink.of(M.empty),
+});
+
+export const sink = {
+	...instanceSink,
+	...pipeable(instanceSink),
+	sequenceT: sequenceT(instanceSink),
+	sequenceArray: array.sequence(instanceSink),
+	getSemigroup,
+};