- 001
- 002
- 003
- 004
- 005
- 006
- 007
- 008
- 009
- 010
- 011
- 012
- 013
- 014
- 015
- 016
- 017
- 018
- 019
- 020
- 021
- 022
- 023
- 024
- 025
- 026
- 027
- 028
- 029
- 030
- 031
- 032
- 033
- 034
- 035
- 036
- 037
- 038
- 039
- 040
- 041
- 042
- 043
- 044
- 045
- 046
- 047
- 048
- 049
- 050
- 051
- 052
- 053
- 054
- 055
- 056
- 057
- 058
- 059
- 060
- 061
- 062
- 063
- 064
- 065
- 066
- 067
- 068
- 069
- 070
- 071
- 072
- 073
- 074
- 075
- 076
- 077
- 078
- 079
- 080
- 081
- 082
- 083
- 084
- 085
- 086
- 087
- 088
- 089
- 090
- 091
- 092
- 093
- 094
- 095
- 096
- 097
- 098
- 099
- 100
package com.example
import kotlinx.coroutines.*
import io.ktor.network.selector.*
import io.ktor.network.sockets.*
import io.ktor.utils.io.*
import kotlinx.coroutines.channels.BroadcastChannel
import kotlinx.coroutines.channels.ClosedReceiveChannelException
import kotlinx.coroutines.channels.ConflatedBroadcastChannel
import kotlinx.coroutines.channels.ReceiveChannel
import java.io.IOException
import java.lang.StringBuilder
import java.nio.ByteBuffer
suspend fun ByteReadChannel.readString(): String {
val result = StringBuilder()
val decoder = Charsets.US_ASCII.newDecoder()
val buffer = ByteBuffer.allocate(1)
while (!isClosedForRead) {
val byte = readByte()
if (byte > 127 || byte < 0) {
continue
}
val c = decoder.decode(buffer.also {
it.put(byte)
it.rewind()
})[0]
result.append(c)
if (c == '\n') {
return result.toString().trim('\r', '\n')
}
buffer.rewind()
decoder.reset()
}
return ""
}
suspend fun ByteWriteChannel.println(text: String) {
writeStringUtf8(text)
writeStringUtf8("\r\n")
}
class Client(private val clientSocket: Socket, private val room: BroadcastChannel<String>) {
private val output = clientSocket.openWriteChannel(autoFlush = true)
private val input = clientSocket.openReadChannel()
var nick: String? = null
private set
suspend fun start() = coroutineScope {
input.discard(input.availableForRead.toLong())
output.writeStringUtf8("Welcome! And your name: ")
val nick = input.readString()
room.send("$nick is here")
output.println("Welcome $nick")
[email protected] = nick
val roomSubscription = room.openSubscription()
launch {
for (message in roomSubscription) {
output.println(message)
}
}
launch {
processUserInput(nick)
}.join()
roomSubscription.cancel()
}
private suspend fun processUserInput(nick: String) {
while (!clientSocket.isClosed) {
val text = input.readString()
room.send("$nick: $text")
if (text == "bye") {
room.send("$nick left")
return
}
}
}
}
suspend fun stdoutRoomProcessor(input: ReceiveChannel<String>) {
for (message in input) {
println(message)
}
}
suspend fun server(port: Int) = coroutineScope {
val serverSocket = aSocket(ActorSelectorManager(coroutineContext)).tcp().bind(port = port)
val room = ConflatedBroadcastChannel<String>()
launch {
stdoutRoomProcessor(room.openSubscription())
}
while (coroutineContext.isActive) {
val clientSocket = serverSocket.accept()
room.send("Client connected ${clientSocket.remoteAddress}")
launch {
val client = Client(clientSocket, room)
try {
client.start()
Follow us!