jcornaz / miop Goto Github PK

fun <A> StateStore<", A>.dispatch(initialAction: A, onError: (Throwable) -> A, finalAction: suspend () -> A) {
	dispatch(initialAction)
	launch(Unconfined) {
		try {
			dispatch(finalAction())
		} catch(error: Throwable) {
			dispatch(onError(error))
		}
	}
}

interface AsyncAction<R, out A> {

  /** pure */
  val context: CoroutineContext get() = DefaultDispatcher

  /** pure */
  val onStart: A

  /** pure */
  fun onComplete(result: R): A

  /** pure */
  fun onError(error: Throwable): A

  /**
   * Do not have to be pure. May perform side effect and return different result over invocation.
   */
  suspend fun execute(): R
}

fun <R, A> StateStore<*, A>.dispatch(action: AsyncAction<R, A>) {
  dispatch(action.onStart)
  launch(action.context) {
    try {
      dispatch(action.onComplete(action.execute()))
    } catch (error: Throwable) {
      dispatch(action.onError(error))
    }
  }
}

suspend fun <T, R> ReceiveChannel<T>.transform(
    context: CoroutineContext = Unconfined,
    body: suspend (input: ReceiveChannel<T>, output: SendChannel<R>) -> Unit
) = produce<R>(context, onCompletion = consumes()) {
    body(this@transform, channel)
}

class Barrier private constructor(
    private val lock: Mutex,
    private val actualJob: Deferred<Unit> = async(Unconfined, start = CoroutineStart.LAZY) { lock.withLock { } }
) : Deferred<Unit> by actualJob {
  constructor() : this(Mutex(true))

  fun relase() {
    lock.unlock()
  }
}

fun main(args: Array<String>) = runBlocking<Unit> {
  val barrier = Barrier()

  launch(coroutineContext, start = CoroutineStart.UNDISPATCHED) {
    barrier.await()
    print("Hello ")
  }

  println("world")
  barrier.release()

  // barrier.cancel() can also be used to notify any suspended coroutines that the barrier will never be released.
}

subscribable.openSubscription().map { it.field }.distinctUntilChanged()

fun <T, R> SubscribableValue<T>.openSubscription(context: CoroutineContext = Unconfined, transform: suspend (T) -> R): ReceiveChannel<R> =
    openSubscription().transform(context) { input, output ->
      var previous = input.receive()
      output.send(previous)

      input.consumeEach { element ->
        if (element !== previous) {
          output.send(transform(it))
          previous = element
        }
      }
    }

val originalStore: StateStore<S1, A1> = StateStore(intitialState)

val view: StateStore<S2, A2> = originalStore.map(
  transformState = { it.subState },
  transformAction = { adaptAction(it) },
)

inline fun <reified R> ReceiveChannel<*>.filterIsInstance(): ReceiveChannel<R> = transform { input, output ->
  input.consumeEach {
    if (it is R) output.send(it)
  }
}

emptyReceiveChannel<Int>().distinctUntilChanged().consumeEach { println(it) }

fun <E> switchOnError(logError: (Throwable) -> Unit = {}, openChannel: () -> ReceiveChannel<E>): ReceiveChannel<E> = produce(Unconfined) {
  while(isActive) {
	try {
		openChannel().consumeEach { send(it) }
	} catch(t: Throwable) {
		logError(t)
	}
  }
}

/**
 * Start a job in the JavaFx thread which keeps up-to-date the [target] collection.
 * Order of elements is ignored. Only consider the elements and their occurrence count.
 */
fun <E> ReceiveChannel<Collection<E>>.launchFxCollectionUpdater(target: MutableCollection<E>, parent: Job? = null): Job

fun <T> Iterable<T>.openSubscription(context: CoroutineContext = Unconfined, capacity: Int = 0): ReceiveChannel<T> =
	asSequence().openSubscription(context, capacity)
	
fun <T> Sequence<T>.openSubscription(context: CoroutineContext = Unconfined, capacity: Int = 0): ReceiveChannel<T> =
	produce(context, capacity = capacity) { forEach { send(it) } }

fun StateStore<S, E> : SubscribableValue<S> {
  // other members here

  /** suspend until the event has been handled and return the state resulting of the event */
  suspend fun handle(event: E): S
}