repositories {
    mavenCentral()
}

dependencies {
    //for client
    implementation("io.rsocket.kotlin:rsocket-ktor-client:0.15.4")

    //for server
    implementation("io.rsocket.kotlin:rsocket-ktor-server:0.15.4")
}

//create ktor client
val client = HttpClient {
    install(WebSockets) //rsocket requires websockets plugin installed
    install(RSocketSupport) {
        //configure rSocket connector (all values have defaults)
        connector {
            maxFragmentSize = 1024

            connectionConfig {
                keepAlive = KeepAlive(
                    interval = 30.seconds,
                    maxLifetime = 2.minutes
                )

                //payload for setup frame
                setupPayload {
                    buildPayload {
                        data("""{ "data": "setup" }""")
                    }
                }

                //mime types
                payloadMimeType = PayloadMimeType(
                    data = WellKnownMimeType.ApplicationJson,
                    metadata = WellKnownMimeType.MessageRSocketCompositeMetadata
                )
            }

            //optional acceptor for server requests
            acceptor {
                RSocketRequestHandler {
                    requestResponse { it } //echo request payload
                }
            }
        }
    }
}

//connect to some url
val rSocket: RSocket = client.rSocket("wss://demo.rsocket.io/rsocket")

//request stream
val stream: Flow<Payload> = rSocket.requestStream(
    buildPayload {
        data("""{ "data": "hello world" }""")
    }
)

//take 5 values and print response
stream.take(5).collect { payload: Payload ->
    println(payload.data.readText())
}

//create ktor server
embeddedServer(CIO) {
    install(WebSockets) //rsocket requires websockets plugin installed
    install(RSocketSupport) {
        //configure rSocket server (all values have defaults)

        server {
            maxFragmentSize = 1024

            //install interceptors
            interceptors {
                forConnection(::SomeConnectionInterceptor)
            }
        }
    }
    //configure routing
    routing {
        //configure route `/rsocket`
        rSocket("rsocket") {
            println(config.setupPayload.data.readText()) //print setup payload data

            RSocketRequestHandler {
                //handler for request/response
                requestResponse { request: Payload ->
                    println(request.data.readText()) //print request payload data
                    delay(500) // work emulation
                    buildPayload {
                        data("""{ "data": "Server response" }""")
                    }
                }
                //handler for request/stream      
                requestStream { request: Payload ->
                    println(request.data.readText()) //print request payload data
                    flow {
                        repeat(10) { i ->
                            emit(
                                buildPayload {
                                    data("""{ "data": "Server stream response: $i" }""")
                                }
                            )
                        }
                    }
                }
            }
        }
    }
}.start(true)

dependencies {
    implementation("io.rsocket.kotlin:rsocket-core:0.15.4")

    // TCP ktor client/server transport
    implementation("io.rsocket.kotlin:rsocket-transport-ktor-tcp:0.15.4")

    // WS ktor client transport
    implementation("io.rsocket.kotlin:rsocket-transport-ktor-websocket-client:0.15.4")

    // WS ktor server transport
    implementation("io.rsocket.kotlin:rsocket-transport-ktor-websocket-server:0.15.4")

    // TCP nodeJS client/server transport
    implementation("io.rsocket.kotlin:rsocket-transport-nodejs-tcp:0.15.4")
}

val transport = TcpClientTransport("0.0.0.0", 8080)
val connector = RSocketConnector {
    //configuration goes here
}
val rsocket: RSocket = connector.connect(transport)
//use rsocket to do request
val response = rsocket.requestResponse(buildPayload { data("""{ "data": "hello world" }""") })
println(response.data.readText())

val transport = TcpServerTransport("0.0.0.0", 8080)
val connector = RSocketServer {
    //configuration goes here
}
val server: TcpServer = server.bind(transport) {
    RSocketRequestHandler {
        //handler for request/response
        requestResponse { request: Payload ->
            println(request.data.readText()) //print request payload data
            delay(500) // work emulation
            buildPayload {
                data("""{ "data": "Server response" }""")
            }
        }
    }
}
server.handlerJob.join() //wait for server to finish

//assume we have client
val client: RSocket = TODO()

//and stream
val stream: Flow<Payload> = client.requestStream(Payload("data"))

//now we can use `flowOn` to add request strategy to context of flow
//here we use prefetch strategy which will send requestN for 10 elements, when, there is 5 elements left to collect
//so on call `collect`, requestStream frame with requestN will be sent, and then, after 5 elements will be collected
//new requestN with 5 will be sent, so collect will be smooth 
stream.flowOn(PrefetchStrategy(requestSize = 10, requestOn = 5)).collect { payload: Payload ->
    println(payload.data.readText())
}