[arrow-kt] PolymorphicProgram

PolymorphicProgram.kt

/**
 * Arrow 0.10
 * How to write polymorphic programs.
 * See https://arrow-kt.io/docs/patterns/polymorphic_programs for more details...
 */
package me.jorgecastillo.polymorphicapps.polymorphic

import arrow.Kind
import arrow.core.Option
import arrow.core.left
import arrow.core.right
import arrow.fx.IO
import arrow.fx.rx2.*
import arrow.fx.extensions.io.async.async
import arrow.fx.fix
import arrow.fx.rx2.extensions.flowablek.async.async
import arrow.fx.rx2.extensions.maybek.async.async
import arrow.fx.rx2.extensions.observablek.async.async
import arrow.fx.rx2.extensions.singlek.async.async
import arrow.fx.typeclasses.Async
import arrow.typeclasses.ApplicativeError

data class UserId(val value: String)
data class User(val userId: UserId)
data class Task(val value: String)

sealed class UserLookupError : RuntimeException()
data class UserNotInLocalStorage(val user: User) : UserLookupError()
data class UserNotInRemoteStorage(val user: User) : UserLookupError()
data class UnknownError(val underlying: Throwable) : UserLookupError()

interface DataSource<F> {
    fun allTasksByUser(user: User): Kind<F, List<Task>>
}

class LocalDataSource<F>(A: ApplicativeError<F, Throwable>) : DataSource<F>, ApplicativeError<F, Throwable> by A {
    private val localCache: Map<User, List<Task>> =
        mapOf(User(UserId("user1")) to listOf(Task("LocalTask assigned to user1")))

    override fun allTasksByUser(user: User): Kind<F, List<Task>> =
        Option.fromNullable(localCache[user]).fold(
            { raiseError(UserNotInLocalStorage(user)) },
            { just(it) }
        )
}

class RemoteDataSource<F>(A: Async<F>) : DataSource<F>, Async<F> by A {
    private val internetStorage: Map<User, List<Task>> =
        mapOf(User(UserId("user2")) to listOf(Task("Remote Task assigned to user2")))

    override fun allTasksByUser(user: User): Kind<F, List<Task>> =
        async { cb ->
            //allows you to take values from callbacks and place them back in the context of `F`
            Option.fromNullable(internetStorage[user]).fold(
                { cb(UserNotInRemoteStorage(user).left()) },
                { cb(it.right()) }
            )
        }
}

class TaskRepository<F>(
    private val localDS: DataSource<F>,
    private val remoteDS: RemoteDataSource<F>,
    AE: ApplicativeError<F, Throwable>) : ApplicativeError<F, Throwable> by AE {

    fun allTasksByUser(user: User): Kind<F, List<Task>> =
        localDS.allTasksByUser(user).handleErrorWith {
            when (it) {
                is UserNotInLocalStorage -> remoteDS.allTasksByUser(user)
                else -> raiseError(UnknownError(it))
            }
        }
}

class Module<F>(A: Async<F>) {
    private val localDataSource: LocalDataSource<F> = LocalDataSource(A)
    private val remoteDataSource: RemoteDataSource<F> = RemoteDataSource(A)
    val repository: TaskRepository<F> = TaskRepository(localDataSource, remoteDataSource, A)
}

object test {

    @JvmStatic
    fun main(args: Array<String>): Unit {
        val user1 = User(UserId("user1"))
        val user2 = User(UserId("user2"))
        val user3 = User(UserId("unknown user"))

        val singleModule = Module(SingleK.async())
        singleModule.run {
            repository.allTasksByUser(user1).fix().single.subscribe(::println, ::println)
            repository.allTasksByUser(user2).fix().single.subscribe(::println, ::println)
            repository.allTasksByUser(user3).fix().single.subscribe(::println, ::println)
        }

        val maybeModule = Module(MaybeK.async())
        maybeModule.run {
            repository.allTasksByUser(user1).fix().maybe.subscribe(::println, ::println)
            repository.allTasksByUser(user2).fix().maybe.subscribe(::println, ::println)
            repository.allTasksByUser(user3).fix().maybe.subscribe(::println, ::println)
        }

        val observableModule = Module(ObservableK.async())
        observableModule.run {
            repository.allTasksByUser(user1).fix().observable.subscribe(::println)
            repository.allTasksByUser(user2).fix().observable.subscribe(::println)
            repository.allTasksByUser(user3).fix().observable.subscribe(::println, ::println)
        }

        val flowableModule = Module(FlowableK.async())
        flowableModule.run {
            repository.allTasksByUser(user1).fix().flowable.subscribe(::println)
            repository.allTasksByUser(user2).fix().flowable.subscribe(::println)
            repository.allTasksByUser(user3).fix().flowable.subscribe(::println, ::println)
        }

        val ioModule = Module(IO.async())
        ioModule.run {
            println(repository.allTasksByUser(user1).fix().attempt().unsafeRunSync())
            println(repository.allTasksByUser(user2).fix().attempt().unsafeRunSync())
            println(repository.allTasksByUser(user3).fix().attempt().unsafeRunSync())
        }
    }
}