You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

IrcClientImpl.kt 5.9KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143
  1. package com.dmdirc.ktirc
  2. import com.dmdirc.ktirc.events.*
  3. import com.dmdirc.ktirc.events.handlers.eventHandlers
  4. import com.dmdirc.ktirc.events.mutators.eventMutators
  5. import com.dmdirc.ktirc.io.KtorLineBufferedSocket
  6. import com.dmdirc.ktirc.io.LineBufferedSocket
  7. import com.dmdirc.ktirc.io.MessageHandler
  8. import com.dmdirc.ktirc.io.MessageParser
  9. import com.dmdirc.ktirc.messages.*
  10. import com.dmdirc.ktirc.messages.processors.messageProcessors
  11. import com.dmdirc.ktirc.model.*
  12. import com.dmdirc.ktirc.util.currentTimeProvider
  13. import com.dmdirc.ktirc.util.generateLabel
  14. import com.dmdirc.ktirc.util.logger
  15. import io.ktor.util.KtorExperimentalAPI
  16. import kotlinx.coroutines.*
  17. import kotlinx.coroutines.channels.Channel
  18. import kotlinx.coroutines.channels.map
  19. import kotlinx.coroutines.time.withTimeoutOrNull
  20. import java.time.Duration
  21. import java.util.concurrent.atomic.AtomicBoolean
  22. import java.util.logging.Level
  23. /**
  24. * Concrete implementation of an [IrcClient].
  25. */
  26. // TODO: How should alternative nicknames work?
  27. // TODO: Should IRC Client take a pool of servers and rotate through, or make the caller do that?
  28. internal class IrcClientImpl(private val config: IrcClientConfig) : ExperimentalIrcClient, CoroutineScope {
  29. private val log by logger()
  30. @ExperimentalCoroutinesApi
  31. override val coroutineContext = GlobalScope.newCoroutineContext(Dispatchers.IO)
  32. @ExperimentalCoroutinesApi
  33. @KtorExperimentalAPI
  34. internal var socketFactory: (CoroutineScope, String, Int, Boolean) -> LineBufferedSocket = ::KtorLineBufferedSocket
  35. internal var asyncTimeout = Duration.ofSeconds(20)
  36. override var behaviour = config.behaviour
  37. override val serverState = ServerState(config.profile.nickname, config.server.host, config.sasl)
  38. override val channelState = ChannelStateMap { caseMapping }
  39. override val userState = UserState { caseMapping }
  40. private val messageHandler = MessageHandler(messageProcessors, eventMutators, eventHandlers)
  41. private val messageBuilder = MessageBuilder()
  42. private val parser = MessageParser()
  43. private var socket: LineBufferedSocket? = null
  44. private val connecting = AtomicBoolean(false)
  45. override fun send(message: String) {
  46. socket?.sendChannel?.offer(message.toByteArray()) ?: log.warning { "No send channel for message: $message" }
  47. }
  48. override fun send(tags: Map<MessageTag, String>, command: String, vararg arguments: String) {
  49. maybeEchoMessage(command, arguments)
  50. socket?.sendChannel?.offer(messageBuilder.build(tags, command, arguments))
  51. ?: log.warning { "No send channel for command: $command" }
  52. }
  53. override fun sendAsync(tags: Map<MessageTag, String>, command: String, arguments: Array<String>, matcher: (IrcEvent) -> Boolean) = async {
  54. val label = generateLabel(this@IrcClientImpl)
  55. val channel = Channel<IrcEvent>(1)
  56. if (serverState.asyncResponseState.supportsLabeledResponses) {
  57. serverState.asyncResponseState.pendingResponses[label] = channel to { event -> event.metadata.label == label }
  58. send(tags + (MessageTag.Label to label), command, *arguments)
  59. } else {
  60. serverState.asyncResponseState.pendingResponses[label] = channel to matcher
  61. send(tags, command, *arguments)
  62. }
  63. withTimeoutOrNull(asyncTimeout) {
  64. channel.receive()
  65. }.also { serverState.asyncResponseState.pendingResponses.remove(label) }
  66. }
  67. override fun connect() {
  68. check(!connecting.getAndSet(true))
  69. @Suppress("EXPERIMENTAL_API_USAGE")
  70. with(socketFactory(this, config.server.host, config.server.port, config.server.useTls)) {
  71. socket = this
  72. emitEvent(ServerConnecting(EventMetadata(currentTimeProvider())))
  73. launch {
  74. try {
  75. connect()
  76. emitEvent(ServerConnected(EventMetadata(currentTimeProvider())))
  77. sendCapabilityList()
  78. sendPasswordIfPresent()
  79. sendNickChange(config.profile.nickname)
  80. sendUser(config.profile.username, config.profile.realName)
  81. messageHandler.processMessages(this@IrcClientImpl, receiveChannel.map { parser.parse(it) })
  82. } catch (ex: Exception) {
  83. log.log(Level.SEVERE, ex) { "Error connecting to ${config.server.host}:${config.server.port}" }
  84. emitEvent(ServerConnectionError(EventMetadata(currentTimeProvider()), ex.toConnectionError(), ex.localizedMessage))
  85. }
  86. reset()
  87. emitEvent(ServerDisconnected(EventMetadata(currentTimeProvider())))
  88. }
  89. }
  90. }
  91. override fun disconnect() {
  92. socket?.disconnect()
  93. }
  94. override fun onEvent(handler: (IrcEvent) -> Unit) = messageHandler.addEmitter(handler)
  95. private fun emitEvent(event: IrcEvent) = messageHandler.handleEvent(this, event)
  96. private fun sendPasswordIfPresent() = config.server.password?.let(this::sendPassword)
  97. private fun maybeEchoMessage(command: String, arguments: Array<out String>) {
  98. // TODO: Is this the best place to do it? It'd be nicer to actually build the message and
  99. // reflect the raw line back through all the processors etc.
  100. if (command == "PRIVMSG" && behaviour.alwaysEchoMessages && !serverState.capabilities.enabledCapabilities.contains(Capability.EchoMessages)) {
  101. emitEvent(MessageReceived(
  102. EventMetadata(currentTimeProvider()),
  103. userState[serverState.localNickname]?.details ?: User(serverState.localNickname),
  104. arguments[0],
  105. arguments[1]
  106. ))
  107. }
  108. }
  109. internal fun reset() {
  110. serverState.reset()
  111. channelState.clear()
  112. userState.reset()
  113. socket = null
  114. connecting.set(false)
  115. }
  116. }