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 7.0KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172
  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.LineBufferedSocket
  6. import com.dmdirc.ktirc.io.LineBufferedSocketImpl
  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.RemoveIn
  13. import com.dmdirc.ktirc.util.currentTimeProvider
  14. import com.dmdirc.ktirc.util.generateLabel
  15. import com.dmdirc.ktirc.util.logger
  16. import kotlinx.coroutines.*
  17. import kotlinx.coroutines.channels.Channel
  18. import kotlinx.coroutines.channels.map
  19. import kotlinx.coroutines.sync.Mutex
  20. import java.net.Inet6Address
  21. import java.net.InetAddress
  22. import java.time.Duration
  23. import java.util.logging.Level
  24. /**
  25. * Concrete implementation of an [IrcClient].
  26. */
  27. internal class IrcClientImpl(private val config: IrcClientConfig) : ExperimentalIrcClient, CoroutineScope {
  28. private val log by logger()
  29. @ExperimentalCoroutinesApi
  30. override val coroutineContext = GlobalScope.newCoroutineContext(Dispatchers.IO)
  31. @ExperimentalCoroutinesApi
  32. internal var socketFactory: (CoroutineScope, String, String, Int, Boolean) -> LineBufferedSocket = ::LineBufferedSocketImpl
  33. internal var resolver: (String) -> Collection<ResolveResult> = { host ->
  34. InetAddress.getAllByName(host).map { ResolveResult(it.hostAddress, it is Inet6Address) }
  35. }
  36. internal var asyncTimeout = Duration.ofSeconds(20)
  37. override var behaviour = config.behaviour
  38. override val serverState = ServerState(config.profile.nickname, config.server.host, config.sasl)
  39. override val channelState = ChannelStateMap { caseMapping }
  40. override val userState = UserState { caseMapping }
  41. private val messageHandler = MessageHandler(messageProcessors, eventMutators, eventHandlers)
  42. private val messageBuilder = MessageBuilder()
  43. private val parser = MessageParser()
  44. private var socket: LineBufferedSocket? = null
  45. private val connecting = Mutex(false)
  46. @Deprecated("Use structured send instead", ReplaceWith("send(command, arguments)"))
  47. @RemoveIn("2.0.0")
  48. override fun send(message: String) {
  49. socket?.sendChannel?.offer(message.toByteArray()) ?: log.warning { "No send channel for message: $message" }
  50. }
  51. override fun send(tags: Map<MessageTag, String>, command: String, vararg arguments: String) {
  52. maybeEchoMessage(command, arguments)
  53. socket?.sendChannel?.offer(messageBuilder.build(tags, command, arguments))
  54. ?: log.warning { "No send channel for command: $command" }
  55. }
  56. override fun sendAsync(tags: Map<MessageTag, String>, command: String, arguments: Array<String>, matcher: (IrcEvent) -> Boolean) = async {
  57. val label = generateLabel(this@IrcClientImpl)
  58. val channel = Channel<IrcEvent>(1)
  59. if (serverState.asyncResponseState.supportsLabeledResponses) {
  60. serverState.asyncResponseState.pendingResponses[label] = channel to { event -> event.metadata.label == label }
  61. send(tags + (MessageTag.Label to label), command, *arguments)
  62. } else {
  63. serverState.asyncResponseState.pendingResponses[label] = channel to matcher
  64. send(tags, command, *arguments)
  65. }
  66. withTimeoutOrNull(asyncTimeout.toMillis()) {
  67. channel.receive()
  68. }.also { serverState.asyncResponseState.pendingResponses.remove(label) }
  69. }
  70. override fun connect() {
  71. check(connecting.tryLock()) { "IrcClient is already connected to a server" }
  72. val ip: String
  73. try {
  74. ip = resolve(config.server.host)
  75. } catch (ex: Exception) {
  76. log.log(Level.SEVERE, ex) { "Error resolving ${config.server.host}" }
  77. emitEvent(ServerConnectionError(EventMetadata(currentTimeProvider()), ConnectionError.UnresolvableAddress, ex.localizedMessage))
  78. reset()
  79. return
  80. }
  81. @Suppress("EXPERIMENTAL_API_USAGE")
  82. with(socketFactory(this, config.server.host, ip, config.server.port, config.server.useTls)) {
  83. socket = this
  84. emitEvent(ServerConnecting(EventMetadata(currentTimeProvider())))
  85. launch {
  86. try {
  87. connect()
  88. emitEvent(ServerConnected(EventMetadata(currentTimeProvider())))
  89. sendCapabilityList()
  90. sendPasswordIfPresent()
  91. sendNickChange(config.profile.nickname)
  92. sendUser(config.profile.username, config.profile.realName)
  93. messageHandler.processMessages(this@IrcClientImpl, receiveChannel.map { parser.parse(it) })
  94. } catch (ex: Exception) {
  95. log.log(Level.SEVERE, ex) { "Error connecting to ${config.server.host}:${config.server.port}" }
  96. emitEvent(ServerConnectionError(EventMetadata(currentTimeProvider()), ex.toConnectionError(), ex.localizedMessage))
  97. }
  98. reset()
  99. emitEvent(ServerDisconnected(EventMetadata(currentTimeProvider())))
  100. }
  101. }
  102. }
  103. override fun disconnect() {
  104. runBlocking {
  105. socket?.disconnect()
  106. connecting.lock()
  107. connecting.unlock()
  108. }
  109. }
  110. override fun onEvent(handler: (IrcEvent) -> Unit) = messageHandler.addEmitter(handler)
  111. private fun emitEvent(event: IrcEvent) = messageHandler.handleEvent(this, event)
  112. private fun sendPasswordIfPresent() = config.server.password?.let(this::sendPassword)
  113. private fun maybeEchoMessage(command: String, arguments: Array<out String>) {
  114. // TODO: Is this the best place to do it? It'd be nicer to actually build the message and
  115. // reflect the raw line back through all the processors etc.
  116. if (command == "PRIVMSG" && behaviour.alwaysEchoMessages && !serverState.capabilities.enabledCapabilities.contains(Capability.EchoMessages)) {
  117. emitEvent(MessageReceived(
  118. EventMetadata(currentTimeProvider()),
  119. userState[serverState.localNickname]?.details ?: User(serverState.localNickname),
  120. arguments[0],
  121. arguments[1]
  122. ))
  123. }
  124. }
  125. private fun resolve(host: String): String {
  126. val hosts = resolver(host)
  127. val preferredHosts = hosts.filter { it.isV6 == behaviour.preferIPv6 }
  128. return if (preferredHosts.isNotEmpty()) {
  129. preferredHosts.random().ip
  130. } else {
  131. hosts.random().ip
  132. }
  133. }
  134. internal fun reset() {
  135. serverState.reset()
  136. channelState.clear()
  137. userState.reset()
  138. socket = null
  139. connecting.unlock()
  140. }
  141. }
  142. internal data class ResolveResult(val ip: String, val isV6: Boolean)