diff --git a/.circleci/config.yml b/.circleci/config.yml index c4a9a858..94ea03a6 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -3,7 +3,7 @@ version: 2 jobs: build: docker: - - image: circleci/android:api-28-alpha + - image: circleci/android:api-29 environment: TERM: dumb JAVA_TOOL_OPTIONS: "-Xmx1g" diff --git a/build.gradle b/build.gradle index d8021700..eff08c4a 100644 --- a/build.gradle +++ b/build.gradle @@ -1,7 +1,7 @@ // Top-level build file where you can add configuration options common to all sub-projects/modules. buildscript { - ext.kotlin_version = '1.3.50' + ext.kotlin_version = '1.3.61' ext.dokka_version = '0.9.16' repositories { mavenCentral() @@ -10,7 +10,7 @@ buildscript { } dependencies { classpath 'com.vanniktech:gradle-maven-publish-plugin:0.8.0' - classpath 'com.android.tools.build:gradle:3.4.2' + classpath 'com.android.tools.build:gradle:3.5.3' classpath "org.jetbrains.kotlin:kotlin-gradle-plugin:$kotlin_version" } } diff --git a/demo/build.gradle b/demo/build.gradle index bd5ea465..6ddcad62 100755 --- a/demo/build.gradle +++ b/demo/build.gradle @@ -23,6 +23,10 @@ android { proguardFiles getDefaultProguardFile('proguard-android.txt'), 'proguard-rules.pro' } } + compileOptions { + sourceCompatibility JavaVersion.VERSION_1_8 + targetCompatibility JavaVersion.VERSION_1_8 + } packagingOptions { exclude 'META-INF/rxjava.properties' exclude 'META-INF/ASL-2.0.txt' diff --git a/demo/src/main/java/com/tinder/app/root/view/DemoActivity.kt b/demo/src/main/java/com/tinder/app/root/view/DemoActivity.kt index 6d231fa5..fd54f9d0 100755 --- a/demo/src/main/java/com/tinder/app/root/view/DemoActivity.kt +++ b/demo/src/main/java/com/tinder/app/root/view/DemoActivity.kt @@ -41,9 +41,12 @@ class DemoActivity : AppCompatActivity() { tabLayout.addOnTabSelectedListener(TabLayout.ViewPagerOnTabSelectedListener(viewPager)) } - private class ViewPagerAdapter constructor(fm: FragmentManager, private val pageCount: Int) : - FragmentStatePagerAdapter(fm) { - override fun getItem(position: Int): Fragment? { + private class ViewPagerAdapter( + fm: FragmentManager, + private val pageCount: Int + ) : FragmentStatePagerAdapter(fm, BEHAVIOR_RESUME_ONLY_CURRENT_FRAGMENT) { + + override fun getItem(position: Int): Fragment { val (_, createFragment) = TAB_ITEMS[position] return createFragment() } diff --git a/dependencies.gradle b/dependencies.gradle index db1a4143..9bc6d880 100644 --- a/dependencies.gradle +++ b/dependencies.gradle @@ -1,11 +1,11 @@ ext { minSdkVersion = 24 minSdkVersionAndroidLifecycle = 19 - targetSdkVersion = 28 - compileSdkVersion = 28 - buildToolsVersion = '28.0.3' + targetSdkVersion = 29 + compileSdkVersion = 29 + buildToolsVersion = '29.0.0' - appCompat = 'androidx.appcompat:appcompat:1.0.0' + appCompat = 'androidx.appcompat:appcompat:1.1.0' lifecycleExtensions = 'androidx.lifecycle:lifecycle-extensions:2.0.0' material = 'com.google.android.material:material:1.0.0' multiDex = 'androidx.multidex:multidex:2.0.0' @@ -17,16 +17,22 @@ ext { rxJava3 = 'io.reactivex.rxjava3:rxkotlin:3.0.0' rxAndroid3 = 'io.reactivex.rxjava3:rxandroid:3.0.0' rxKotlin3 = 'io.reactivex.rxjava3:rxkotlin:3.0.0' - rxJava = 'io.reactivex.rxjava2:rxjava:2.1.16' - rxAndroid = 'io.reactivex.rxjava2:rxandroid:2.0.2' - rxKotlin = 'io.reactivex.rxjava2:rxkotlin:2.2.0' + + def rxJavaVersion = "2.2.15" + def rxAndroidVersion = "2.1.1" + def rxKotlinVersion = "2.4.0" + rxJava = "io.reactivex.rxjava2:rxjava:$rxJavaVersion" + rxAndroid = "io.reactivex.rxjava2:rxandroid:$rxAndroidVersion" + rxKotlin = "io.reactivex.rxjava2:rxkotlin:$rxKotlinVersion" rxJava1 = 'io.reactivex:rxjava:1.3.4' + kotlinCoroutines = 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.2' kotlinCoroutinesRxInterop = 'org.jetbrains.kotlinx:kotlinx-coroutines-reactive:1.3.2' kotlinCoroutinesTest = 'org.jetbrains.kotlinx:kotlinx-coroutines-test:1.3.2' - stetho = 'com.facebook.stetho:stetho:1.5.0' - stethoOkHttp = 'com.facebook.stetho:stetho-okhttp3:1.5.0' + def stethoVersion = "1.5.0" + stetho = "com.facebook.stetho:stetho:$stethoVersion" + stethoOkHttp = "com.facebook.stetho:stetho-okhttp3:$stethoVersion" moshi = 'com.squareup.moshi:moshi-kotlin:1.5.0' moshiCodeGen = 'com.squareup.moshi:moshi-kotlin-codegen:1.6.0' @@ -44,20 +50,23 @@ ext { jodaTime = 'joda-time:joda-time:2.9.9' glide = 'com.github.bumptech.glide:glide:4.5.0' - okHttp = 'com.squareup.okhttp3:okhttp:3.11.0' - okHttpServerSentEvent = 'com.squareup.okhttp3:okhttp-sse:3.11.0' - okHttpLoggingInterceptor = 'com.squareup.okhttp3:logging-interceptor:3.11.0' - okio = 'com.squareup.okio:okio:1.13.0' + def okHttpVersion = "3.12.7" + okHttp = "com.squareup.okhttp3:okhttp:$okHttpVersion" + okHttpServerSentEvent = "com.squareup.okhttp3:okhttp-sse:$okHttpVersion" + okHttpLoggingInterceptor = "com.squareup.okhttp3:logging-interceptor:$okHttpVersion" + + okio = 'com.squareup.okio:okio:2.5.0' mockWebServer = 'com.squareup.okhttp3:mockwebserver:3.11.0' timber = 'com.jakewharton.timber:timber:4.6.0' okSse = 'com.github.heremaps:oksse:0.9.0' socketIoClient = 'io.socket:socket.io-client:1.0.0' nettySocketIoServer = 'com.corundumstudio.socketio:netty-socketio:1.7.12' pahoMqtt = 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.0' - activeMqJUnit = 'org.apache.activemq.tooling:activemq-junit:5.13.1' - activeMqBroker = 'org.apache.activemq:activemq-broker:5.15.6' - activeMqStomp = 'org.apache.activemq:activemq-stomp:5.15.6' - activeMqMqtt = 'org.apache.activemq:activemq-mqtt:5.15.6' + activeMqJUnit = 'org.apache.activemq.tooling:activemq-junit:5.15.11' + activeMqBroker = 'org.apache.activemq:activemq-broker:5.15.11' + activeMqStomp = 'org.apache.activemq:activemq-stomp:5.15.11' + activeMqMqtt = 'org.apache.activemq:activemq-mqtt:5.15.11' + activeHttp = 'org.apache.activemq:activemq-http:5.15.11' slf4jSimple = 'org.slf4j:slf4j-simple:1.7.25' diff --git a/scarlet-message-adapter-jackson/src/test/java/com/tinder/scarlet/messageadapter/jackson/JacksonMessageAdapterTest.kt b/scarlet-message-adapter-jackson/src/test/java/com/tinder/scarlet/messageadapter/jackson/JacksonMessageAdapterTest.kt index 52d7beda..a2357400 100644 --- a/scarlet-message-adapter-jackson/src/test/java/com/tinder/scarlet/messageadapter/jackson/JacksonMessageAdapterTest.kt +++ b/scarlet-message-adapter-jackson/src/test/java/com/tinder/scarlet/messageadapter/jackson/JacksonMessageAdapterTest.kt @@ -58,7 +58,7 @@ internal class JacksonMessageAdapterTest { any(), any().containingText(expectedString) ) - serverAnImplementationObserver.awaitCount(1) + serverAnImplementationObserver.awaitCountAndCheck(1) assertThat(serverAnImplementationObserver.values).containsExactly(data) } @@ -79,7 +79,7 @@ internal class JacksonMessageAdapterTest { any(), any().containingText(expectedString) ) - serverAnImplementationObserver.awaitCount(1) + serverAnImplementationObserver.awaitCountAndCheck(1) assertThat(serverAnImplementationObserver.values).containsExactly(data) } diff --git a/scarlet-protocol-mqtt/src/test/java/com/tinder/scarlet/mqtt/PahoMqttClientIntegrationTest.kt b/scarlet-protocol-mqtt/src/test/java/com/tinder/scarlet/mqtt/PahoMqttClientIntegrationTest.kt index 213b02de..1f2b51a3 100644 --- a/scarlet-protocol-mqtt/src/test/java/com/tinder/scarlet/mqtt/PahoMqttClientIntegrationTest.kt +++ b/scarlet-protocol-mqtt/src/test/java/com/tinder/scarlet/mqtt/PahoMqttClientIntegrationTest.kt @@ -55,7 +55,7 @@ class PahoMqttClientIntegrationTest { connection1.clientClosure() LOGGER.info("${queueTextObserver.values}") - queueTextObserver.awaitCount(2) + queueTextObserver.awaitCountAndCheck(2) } companion object { diff --git a/scarlet-protocol-stomp/build.gradle b/scarlet-protocol-stomp/build.gradle index 3eb282f1..3962324b 100644 --- a/scarlet-protocol-stomp/build.gradle +++ b/scarlet-protocol-stomp/build.gradle @@ -12,8 +12,11 @@ dependencies { implementation project(':scarlet-core-internal') implementation rootProject.ext.rxJava + implementation rootProject.ext.okio implementation rootProject.ext.kotlinStdlib + api rootProject.ext.okHttp + testImplementation project(':scarlet-test-utils') testImplementation rootProject.ext.junit testImplementation rootProject.ext.mockito @@ -23,5 +26,8 @@ dependencies { testImplementation rootProject.ext.activeMqJUnit testImplementation rootProject.ext.activeMqBroker testImplementation rootProject.ext.activeMqStomp + testImplementation rootProject.ext.activeHttp testImplementation rootProject.ext.slf4jSimple + + testImplementation "javax.servlet:javax.servlet-api:4.0.1" } diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/StompEvent.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/StompEvent.kt deleted file mode 100644 index 477bfb90..00000000 --- a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/StompEvent.kt +++ /dev/null @@ -1,9 +0,0 @@ -/* - * © 2018 Match Group, LLC. - */ - -package com.tinder.scarlet.stomp - -import com.tinder.scarlet.ProtocolSpecificEvent - -sealed class StompEvent : ProtocolSpecificEvent diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/OkHttpStompClient.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/OkHttpStompClient.kt new file mode 100644 index 00000000..3bb28fbc --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/OkHttpStompClient.kt @@ -0,0 +1,53 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.okhttp.client + +import com.tinder.scarlet.Channel +import com.tinder.scarlet.Protocol +import com.tinder.scarlet.ProtocolSpecificEventAdapter +import com.tinder.scarlet.stomp.okhttp.core.IdGenerator +import com.tinder.scarlet.stomp.okhttp.core.WebSocketFactory +import com.tinder.scarlet.stomp.okhttp.generator.UuidGenerator +import com.tinder.scarlet.utils.SimpleProtocolOpenRequestFactory +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.WebSocketListener + +/** + * Scarlet protocol implementation for create StompMainChannel + * @see OkHttpStompMainChannel + */ +class OkHttpStompClient( + private val configuration: OkHttpStompMainChannel.Configuration, + private val okHttpClient: OkHttpClient, + private val requestFactory: (Channel) -> ClientOpenRequest, + private val idGenerator: IdGenerator = UuidGenerator() +) : Protocol { + + override fun createChannelFactory() = OkHttpStompMainChannel.Factory( + idGenerator = idGenerator, + configuration = configuration, + webSocketFactory = object : WebSocketFactory { + override fun createWebSocket(request: Request, listener: WebSocketListener) { + okHttpClient.newWebSocket(request, listener) + } + } + ) + + override fun createOpenRequestFactory( + channel: Channel + ) = SimpleProtocolOpenRequestFactory { + requestFactory.invoke(channel) + } + + override fun createEventAdapterFactory(): ProtocolSpecificEventAdapter.Factory { + return object : ProtocolSpecificEventAdapter.Factory {} + } + + data class ClientOpenRequest( + val okHttpRequest: Request, + val login: String? = null, + val passcode: String? = null + ) : Protocol.OpenRequest +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/OkHttpStompDestination.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/OkHttpStompDestination.kt new file mode 100644 index 00000000..ad91b263 --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/OkHttpStompDestination.kt @@ -0,0 +1,69 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.okhttp.client + +import com.tinder.scarlet.Channel +import com.tinder.scarlet.Message +import com.tinder.scarlet.Protocol +import com.tinder.scarlet.ProtocolSpecificEventAdapter +import com.tinder.scarlet.stomp.okhttp.models.StompHeader +import com.tinder.scarlet.utils.SimpleChannelFactory +import com.tinder.scarlet.utils.SimpleProtocolOpenRequestFactory + +private typealias DestinationOpenRequestHeaderFactory = (Channel) -> OkHttpStompDestination.DestinationOpenRequest +private typealias MessageMetaDataFactory = (Channel, Message) -> OkHttpStompDestination.MessageMetaData + +/** + * Scarlet protocol implementation for create channel (OkHttpStompMessageChannel) for subscribe to + * queue by destination. + * @see OkHttpStompMessageChannel + * + * MessageMetaDataFactory is optional factory for create custom header for each message which will be sent + * by this StompMessageChannel. + * + * DestinationOpenRequestHeaderFactory is optional factory for create open request header which will be sent + * with subscribe message. + */ +class OkHttpStompDestination( + private val destination: String, + private val openRequestFactory: DestinationOpenRequestHeaderFactory? = null, + private val createMessageMetaDataCallable: MessageMetaDataFactory? = null +) : Protocol { + + override fun createChannelFactory() = SimpleChannelFactory { listener, parent -> + require(parent is OkHttpStompMainChannel) + OkHttpStompMessageChannel(destination, parent, parent, listener) + } + + override fun createOpenRequestFactory(channel: Channel) = SimpleProtocolOpenRequestFactory { + openRequestFactory?.invoke(channel) ?: Protocol.OpenRequest.Empty + } + + override fun createOutgoingMessageMetaDataFactory( + channel: Channel + ) = SimpleMessageMetaDataFactory(createMessageMetaDataCallable) + + override fun createEventAdapterFactory(): ProtocolSpecificEventAdapter.Factory { + return object : ProtocolSpecificEventAdapter.Factory {} + } + + class SimpleMessageMetaDataFactory( + private val createMessageMetaDataCallable: MessageMetaDataFactory? + ) : Protocol.MessageMetaData.Factory { + + override fun create( + channel: Channel, + message: Message + ) = createMessageMetaDataCallable?.invoke(channel, message) + ?: Protocol.MessageMetaData.Empty + } + + data class MessageMetaData( + val headers: StompHeader + ) : Protocol.MessageMetaData + + data class DestinationOpenRequest( + val headers: StompHeader + ) : Protocol.OpenRequest +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/OkHttpStompMainChannel.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/OkHttpStompMainChannel.kt new file mode 100644 index 00000000..db185531 --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/OkHttpStompMainChannel.kt @@ -0,0 +1,308 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.okhttp.client + +import com.tinder.scarlet.Channel +import com.tinder.scarlet.Protocol +import com.tinder.scarlet.stomp.okhttp.core.Connection +import com.tinder.scarlet.stomp.okhttp.core.IdGenerator +import com.tinder.scarlet.stomp.okhttp.core.MessageHandler +import com.tinder.scarlet.stomp.okhttp.core.StompListener +import com.tinder.scarlet.stomp.okhttp.core.StompSender +import com.tinder.scarlet.stomp.okhttp.core.StompSubscriber +import com.tinder.scarlet.stomp.okhttp.core.WebSocketFactory +import com.tinder.scarlet.stomp.okhttp.models.StompCommand +import com.tinder.scarlet.stomp.okhttp.models.StompHeader +import com.tinder.scarlet.stomp.okhttp.models.StompMessage +import com.tinder.scarlet.stomp.okhttp.support.StompHeaderAccessor +import okhttp3.Response +import okhttp3.WebSocket +import okhttp3.WebSocketListener +import okio.ByteString +import java.util.concurrent.ConcurrentHashMap +import kotlin.math.max + +/** + * The main channel, which is responsible for connecting and disconnecting to the stomp server. + * And also for sending messages and the logic of subscriptions. + */ +class OkHttpStompMainChannel( + private val configuration: Configuration, + private val idGenerator: IdGenerator, + private val webSocketFactory: WebSocketFactory, + private val listener: Channel.Listener +) : Channel, StompSender, StompSubscriber { + + companion object { + + /** STOMP recommended error of margin for receiving heartbeats. */ + private const val HEARTBEAT_MULTIPLIER = 3 + + private const val ACCEPT_VERSION = "1.1,1.2" + } + + private val topicIds = ConcurrentHashMap() + private val subscriptions = ConcurrentHashMap() + + private var messageHandler: MessageHandler? = null + private var connection: Connection? = null + + override fun open(openRequest: Protocol.OpenRequest) { + val clientOpenRequest = openRequest as OkHttpStompClient.ClientOpenRequest + + webSocketFactory.createWebSocket( + clientOpenRequest.okHttpRequest, + InnerWebSocketListener(clientOpenRequest) + ) + } + + override fun forceClose() { + disconnect() + connection?.forceClose() + + connection = null + messageHandler = null + } + + override fun close(closeRequest: Protocol.CloseRequest) { + disconnect() + connection?.close() + + connection = null + messageHandler = null + } + + override fun convertAndSend( + payload: ByteArray, + destination: String, + headers: StompHeader? + ): Boolean { + val stompHeaders = StompHeaderAccessor.of(headers.orEmpty()) + .apply { this.destination = destination } + .createHeader() + + val stompMessage = StompMessage.Builder() + .withPayload(payload) + .withHeaders(stompHeaders) + .create(StompCommand.SEND) + + return connection?.sendMessage(stompMessage) ?: false + } + + override fun subscribe( + destination: String, + headers: StompHeader?, + listener: StompListener + ) { + check(!topicIds.containsKey(destination)) { "Already has subscription to destination=$destination" } + check(!subscriptions.containsKey(destination)) { "Already has subscription to destination=$destination" } + val generateId = idGenerator.generateId() + val stompHeaders = StompHeaderAccessor.of(headers.orEmpty()) + .apply { + this.subscriptionId = generateId + this.destination = destination + } + .createHeader() + + val stompMessage = StompMessage.Builder() + .withHeaders(stompHeaders) + .create(StompCommand.SUBSCRIBE) + + connection?.sendMessage(stompMessage) + + topicIds[destination] = generateId + subscriptions[destination] = listener + } + + override fun unsubscribe(destination: String) { + val subscriptionId = topicIds.remove(destination) ?: return + + val stompHeaders = StompHeaderAccessor.of() + .apply { this.subscriptionId = subscriptionId } + .createHeader() + + val stompMessage = StompMessage.Builder() + .withHeaders(stompHeaders) + .create(StompCommand.UNSUBSCRIBE) + + connection?.sendMessage(stompMessage) + subscriptions.remove(destination) + } + + private fun handleIncome(stompMessage: StompMessage) = when (stompMessage.command) { + StompCommand.CONNECTED -> { + setupHeartBeat(stompMessage) + listener.onOpened(this) + } + StompCommand.MESSAGE -> stompMessage.headers.destination + ?.let { destination -> + val listener = subscriptions[destination] + listener?.invoke(stompMessage) + } + StompCommand.ERROR -> listener.onFailed( + this, + configuration.shouldRetryAfterError, + null + ) + else -> Unit // not a server message + } + + private fun setupHeartBeat(stompMessage: StompMessage) { + val (serverSendInterval, serverReceiveInterval) = stompMessage.headers.heartBeat + + val clientSendInterval = configuration.heartbeatSendInterval + val clientReceiveInterval = configuration.heartbeatReceiveInterval + + if (clientSendInterval > 0 && serverReceiveInterval > 0) { + val interval = max(clientSendInterval, serverReceiveInterval) + connection?.onWriteInactivity(interval, ::sendHeartBeat) + } + + if (clientReceiveInterval > 0 && serverSendInterval > 0) { + val interval = max(clientReceiveInterval, serverSendInterval) * HEARTBEAT_MULTIPLIER + connection?.onReceiveInactivity(interval) { + sendErrorMessage("No messages received in $interval ms.") + connection?.close() + listener.onFailed(this@OkHttpStompMainChannel, true, null) + } + } + } + + private fun sendHeartBeat() { + val stompMessage = StompMessage.Builder() + .create(StompCommand.HEARTBEAT) + + connection?.sendMessage(stompMessage) + } + + private fun sendErrorMessage(error: String) { + val headers = StompHeaderAccessor.of() + .apply { message(error) } + .createHeader() + + val stompMessage = StompMessage.Builder() + .withHeaders(headers) + .create(StompCommand.ERROR) + + connection?.sendMessage(stompMessage) + } + + private inner class InnerWebSocketListener( + private val openRequest: OkHttpStompClient.ClientOpenRequest + ) : WebSocketListener() { + + override fun onOpen(webSocket: WebSocket, response: Response) { + val webSocketConnection = WebSocketConnection(webSocket) + + this@OkHttpStompMainChannel.connection = webSocketConnection + this@OkHttpStompMainChannel.messageHandler = webSocketConnection + + val host = configuration.host + val login = openRequest.login + val passcode = openRequest.passcode + + sendConnectMessage(host, login, passcode) + } + + override fun onMessage(webSocket: WebSocket, bytes: ByteString) { + messageHandler?.handle(bytes.toByteArray())?.let(::handleIncome) + } + + override fun onMessage(webSocket: WebSocket, text: String) { + messageHandler?.handle(text.toByteArray(Charsets.UTF_8))?.let(::handleIncome) + } + + override fun onClosing(webSocket: WebSocket, code: Int, reason: String) { + if (WebSocketCode.isUnexpectedClose(code)) { + listener.onFailed(this@OkHttpStompMainChannel, true, null) + this@OkHttpStompMainChannel.connection = null + } else { + listener.onClosing(this@OkHttpStompMainChannel) + } + } + + override fun onClosed(webSocket: WebSocket, code: Int, reason: String) { + listener.onClosed(this@OkHttpStompMainChannel) + this@OkHttpStompMainChannel.connection = null + } + + override fun onFailure(webSocket: WebSocket, throwable: Throwable, response: Response?) { + listener.onFailed( + this@OkHttpStompMainChannel, + configuration.shouldRetryAfterError, + throwable + ) + this@OkHttpStompMainChannel.connection = null + } + } + + private fun sendConnectMessage(host: String, login: String? = null, passcode: String? = null) { + val stompHeaderAccessor = StompHeaderAccessor.of() + .apply { + this.host = host + this.acceptVersion = ACCEPT_VERSION + this.login = login + this.passcode = passcode + } + + if (configuration.shouldSendHeartBeat) { + stompHeaderAccessor.heartBeat = configuration.headerBeatPair + } + + val stompMessage = StompMessage.Builder() + .withHeaders(stompHeaderAccessor.createHeader()) + .create(StompCommand.CONNECT) + + connection?.sendMessage(stompMessage) + } + + private fun disconnect() { + subscriptions.keys.forEach(::unsubscribe) + + topicIds.clear() + subscriptions.clear() + + sendDisconnectMessage() + } + + private fun sendDisconnectMessage() { + val stompMessage = StompMessage.Builder() + .create(StompCommand.DISCONNECT) + + connection?.sendMessage(stompMessage) + } + + data class Configuration( + val host: String, + val shouldRetryAfterError: Boolean = true, + val heartbeatSendInterval: Long = 0, + val heartbeatReceiveInterval: Long = 0 + ) { + + val shouldSendHeartBeat: Boolean + get() = heartbeatSendInterval > 0 && heartbeatReceiveInterval > 0 + + val headerBeatPair: Pair + get() = heartbeatSendInterval to heartbeatReceiveInterval + } + + class Factory( + private val idGenerator: IdGenerator, + private val configuration: Configuration, + private val webSocketFactory: WebSocketFactory + ) : Channel.Factory { + + override fun create( + listener: Channel.Listener, + parent: Channel? + ): Channel? { + return OkHttpStompMainChannel( + configuration = configuration, + idGenerator = idGenerator, + webSocketFactory = webSocketFactory, + listener = listener + ) + } + } +} diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/OkHttpStompMessageChannel.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/OkHttpStompMessageChannel.kt new file mode 100644 index 00000000..3cef1e63 --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/OkHttpStompMessageChannel.kt @@ -0,0 +1,67 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.okhttp.client + +import com.tinder.scarlet.Channel +import com.tinder.scarlet.Message +import com.tinder.scarlet.MessageQueue +import com.tinder.scarlet.Protocol +import com.tinder.scarlet.stomp.okhttp.core.StompSender +import com.tinder.scarlet.stomp.okhttp.core.StompSubscriber + +/** + * This channel is responsible for a specific subscription and its life cycle. + */ +class OkHttpStompMessageChannel( + private val destination: String, + private val stompSubscriber: StompSubscriber, + private val stompSender: StompSender, + private val listener: Channel.Listener +) : Channel, MessageQueue { + + private var messageQueueListener: MessageQueue.Listener? = null + + override fun open(openRequest: Protocol.OpenRequest) { + val destinationOpenRequest = openRequest as? OkHttpStompDestination.DestinationOpenRequest + val stompHeaders = destinationOpenRequest?.headers + stompSubscriber.subscribe(destination, stompHeaders) { message -> + messageQueueListener?.onMessageReceived( + channel = this, + messageQueue = this, + message = Message.Text(message.payload.toString(Charsets.UTF_8)), + metadata = OkHttpStompDestination.MessageMetaData(message.headers) + ) + } + listener.onOpened(this) + } + + override fun close(closeRequest: Protocol.CloseRequest) { + forceClose() + } + + override fun forceClose() { + stompSubscriber.unsubscribe(destination) + listener.onClosed(this) + } + + override fun createMessageQueue(listener: MessageQueue.Listener): MessageQueue { + require(messageQueueListener == null) { "message queue was already created" } + messageQueueListener = listener + return this + } + + override fun send( + message: Message, + messageMetaData: Protocol.MessageMetaData + ): Boolean = when (message) { + is Message.Text -> { + val metaData = messageMetaData as? OkHttpStompDestination.MessageMetaData + stompSender.convertAndSend(message.value.toByteArray(), destination, metaData?.headers) + } + is Message.Bytes -> { + val metaData = messageMetaData as? OkHttpStompDestination.MessageMetaData + stompSender.convertAndSend(message.value, destination, metaData?.headers) + } + } +} diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/WebSocketCode.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/WebSocketCode.kt new file mode 100644 index 00000000..3783cc6f --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/WebSocketCode.kt @@ -0,0 +1,18 @@ +package com.tinder.scarlet.stomp.okhttp.client + +enum class WebSocketCode(val code: Int, val reason: String? = null) { + + CLOSE_NORMAL(1000, "Normal closure"), + CLOSE_GOING_AWAY(1001, "Unexpected closure from the Server"), + CLOSED_NO_STATUS(1005, "Expected close status, received none"), ; + + companion object { + + /** + * @return true is this unexpected situation error code + */ + fun isUnexpectedClose(code: Int): Boolean { + return code == CLOSE_GOING_AWAY.code || code == CLOSED_NO_STATUS.code + } + } +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/WebSocketConnection.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/WebSocketConnection.kt new file mode 100644 index 00000000..49468a60 --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/client/WebSocketConnection.kt @@ -0,0 +1,97 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.okhttp.client + +import com.tinder.scarlet.stomp.okhttp.core.Connection +import com.tinder.scarlet.stomp.okhttp.core.MessageHandler +import com.tinder.scarlet.stomp.okhttp.models.StompMessage +import com.tinder.scarlet.stomp.okhttp.support.StompMessageDecoder +import com.tinder.scarlet.stomp.okhttp.support.StompMessageEncoder +import okhttp3.WebSocket +import java.util.concurrent.Executors +import java.util.concurrent.ScheduledExecutorService +import java.util.concurrent.TimeUnit + +/** + * Okhttp websocket based implementation of {@link Connection}. + */ +class WebSocketConnection( + private val webSocket: WebSocket, + private val executor: ScheduledExecutorService = Executors.newSingleThreadScheduledExecutor() +) : Connection, MessageHandler { + + @Volatile + private var lastReadTime: Long = -1 + + @Volatile + private var lastWriteTime: Long = -1 + + private val messageEncoder = StompMessageEncoder() + private val messageDecoder = StompMessageDecoder() + + /** + * {@inheritDoc} + */ + override fun sendMessage(message: StompMessage): Boolean { + val lastWriteTime = lastWriteTime + if (lastWriteTime != -1L) { + this.lastWriteTime = System.nanoTime() + } + val encodedMessage = messageEncoder.encode(message).toString(Charsets.UTF_8) + return webSocket.send(encodedMessage) + } + + /** + * {@inheritDoc} + */ + override fun onReceiveInactivity(duration: Long, runnable: () -> Unit) { + check(duration > 0) { "Duration must be more than 0" } + lastReadTime = System.nanoTime() + executor.scheduleWithFixedDelay({ + if ((System.nanoTime() - lastReadTime) > TimeUnit.MILLISECONDS.toNanos(duration)) { + runnable.invoke() + } + }, 0, duration / 2L, TimeUnit.MILLISECONDS) + } + + /** + * {@inheritDoc} + */ + override fun onWriteInactivity(duration: Long, runnable: () -> Unit) { + check(duration > 0) { "Duration must be more than 0" } + lastWriteTime = System.nanoTime() + executor.scheduleWithFixedDelay({ + if ((System.nanoTime() - lastWriteTime) > TimeUnit.MILLISECONDS.toNanos(duration)) { + runnable.invoke() + } + }, 0, duration / 2L, TimeUnit.MILLISECONDS) + } + + /** + * {@inheritDoc} + */ + override fun forceClose() { + webSocket.cancel() + executor.shutdown() + } + + /** + * {@inheritDoc} + */ + override fun close() { + webSocket.close(WebSocketCode.CLOSE_NORMAL.code, WebSocketCode.CLOSE_NORMAL.reason) + executor.shutdown() + } + + /** + * {@inheritDoc} + */ + override fun handle(data: ByteArray): StompMessage? { + val lastReadTime = lastReadTime + if (lastReadTime != -1L) { + this.lastReadTime = System.nanoTime() + } + return messageDecoder.decode(data) + } +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/Connection.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/Connection.kt new file mode 100644 index 00000000..92951c13 --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/Connection.kt @@ -0,0 +1,40 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.okhttp.core + +import com.tinder.scarlet.stomp.okhttp.models.StompMessage + +interface Connection { + + /** + * Send the given message. + * @param message the message + * @return true if the message was enqueued. + */ + fun sendMessage(message: StompMessage): Boolean + + /** + * Register a task to invoke after a period of read inactivity. + * @param runnable the task to invoke + * @param duration the amount of inactive time in milliseconds + */ + fun onReceiveInactivity(duration: Long, runnable: () -> Unit) + + /** + * Register a task to invoke after a period of write inactivity. + * @param runnable the task to invoke + * @param duration the amount of inactive time in milliseconds + */ + fun onWriteInactivity(duration: Long, runnable: () -> Unit) + + /** + * Force close the connection. + */ + fun forceClose() + + /** + * Close the connection. + */ + fun close() +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/IdGenerator.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/IdGenerator.kt new file mode 100644 index 00000000..1d696194 --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/IdGenerator.kt @@ -0,0 +1,12 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.okhttp.core + +interface IdGenerator { + + /** + * Generate a new identifier. + */ + fun generateId(): String +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/MessageHandler.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/MessageHandler.kt new file mode 100644 index 00000000..ddf64c00 --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/MessageHandler.kt @@ -0,0 +1,14 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.okhttp.core + +import com.tinder.scarlet.stomp.okhttp.models.StompMessage + +interface MessageHandler { + + /** + * Convert given raw data byte array to stomp message + */ + fun handle(data: ByteArray): StompMessage? +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/StompSender.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/StompSender.kt new file mode 100644 index 00000000..1edd22a4 --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/StompSender.kt @@ -0,0 +1,22 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.okhttp.core + +import com.tinder.scarlet.stomp.okhttp.models.StompHeader + +/** + * Operations for sending messages to a destination. + */ +interface StompSender { + + /** + * Convert the given byte array to serialized form, possibly using a + * StompMessageEncoder, wrap it as a message and send it to a given destination. + * @param payload the byte array to use as payload + * @param destination the target destination + * @param headers headers for the message to send + * @see StompMessageEncoder + */ + fun convertAndSend(payload: ByteArray, destination: String, headers: StompHeader?): Boolean +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/StompSubscriber.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/StompSubscriber.kt new file mode 100644 index 00000000..c2f326a7 --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/StompSubscriber.kt @@ -0,0 +1,25 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.okhttp.core + +import com.tinder.scarlet.stomp.okhttp.models.StompHeader +import com.tinder.scarlet.stomp.okhttp.models.StompMessage + +typealias StompListener = (StompMessage) -> Unit + +/** + * Interface use for subscribe and unsubscribe to STOMP queue. + */ +interface StompSubscriber { + + /** + * Subscribe to given destination with headers. + */ + fun subscribe(destination: String, headers: StompHeader?, listener: StompListener) + + /** + * Unsubscribe from given destination. + */ + fun unsubscribe(destination: String) +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/WebSocketFactory.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/WebSocketFactory.kt new file mode 100644 index 00000000..80b6cacc --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/core/WebSocketFactory.kt @@ -0,0 +1,15 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.okhttp.core + +import okhttp3.Request +import okhttp3.WebSocketListener + +/** + * An factory for create STOMP WebSocket connection. + */ +interface WebSocketFactory { + + fun createWebSocket(request: Request, listener: WebSocketListener) +} diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/generator/UuidGenerator.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/generator/UuidGenerator.kt new file mode 100644 index 00000000..46ee9815 --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/generator/UuidGenerator.kt @@ -0,0 +1,9 @@ +package com.tinder.scarlet.stomp.okhttp.generator + +import com.tinder.scarlet.stomp.okhttp.core.IdGenerator +import java.util.UUID + +class UuidGenerator : IdGenerator { + + override fun generateId() = UUID.randomUUID().toString() +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/models/StompCommand.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/models/StompCommand.kt new file mode 100644 index 00000000..d8251347 --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/models/StompCommand.kt @@ -0,0 +1,27 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.okhttp.models + +/** + * Represents a STOMP command. + */ +enum class StompCommand( + val isBodyAllowed: Boolean = false, + val isDestinationRequired: Boolean = false +) { + // client + CONNECT, + DISCONNECT, + SEND(isBodyAllowed = true, isDestinationRequired = true), + SUBSCRIBE(isDestinationRequired = true), + UNSUBSCRIBE, + + // server + CONNECTED, + MESSAGE(isBodyAllowed = true, isDestinationRequired = true), + ERROR(isBodyAllowed = true), + + // heartbeat + HEARTBEAT +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/models/StompHeader.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/models/StompHeader.kt new file mode 100644 index 00000000..c8527d5d --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/models/StompHeader.kt @@ -0,0 +1,112 @@ +/* + * © 2018 Match Group, LLC. + */ + +package com.tinder.scarlet.stomp.okhttp.models + +/** + * Represents STOMP frame headers. + */ +class StompHeader( + private val headers: Map +) : Map by headers { + + companion object { + + // Standard Headers + /** + * The implied text encoding for MIME types starting with text/ is UTF-8. + * If you are using a text based MIME type with a different encoding then you SHOULD append ;charset= to the MIME type. + * For example, text/html;charset=utf-16 SHOULD be used if your sending an HTML body in UTF-16 encoding. + */ + const val CONTENT_TYPE = "content-type" + /** + * This header is an octet count for the length of the message body. + */ + const val CONTENT_LENGTH = "content-length" + /** + * This will cause the server to acknowledge the processing of the client frame with a RECEIPT frame + */ + const val RECEIPT = "receipt" + /** + * The name of a virtual host that the client wishes to connect to. + */ + const val HOST = "host" + /** + * The versions of the STOMP protocol the client supports. + */ + const val ACCEPT_VERSION = "accept-version" + /** + * The user identifier used to authenticate against a secured STOMP server. + */ + const val LOGIN = "login" + /** + * The password used to authenticate against a secured STOMP server. + */ + const val PASSCODE = "passcode" + /** + * Heart-beating can optionally be used to test the healthiness of the underlying TCP + * connection and to make sure that the remote end is alive and kicking. + * When used, the heart-beat header MUST contain two positive integers separated by a comma. + * + * For detail information see https://stomp.github.io + */ + const val HEARTBEAT = "heart-beat" + /** + * A session identifier that uniquely identifies the session. + */ + const val SESSION = "session" + /** + * A field that contains information about the STOMP server. + * The field MUST contain a server-name field and MAY be followed by optional comment fields delimited by a space character. + */ + const val SERVER = "server" + + /** + * The header indicates where to send the message. + */ + const val DESTINATION = "destination" + /** + * Since a single connection can have multiple open subscriptions with a server, + * an id header MUST be included in the frame to uniquely identify the subscription. + * The id header allows the client and server to relate subsequent MESSAGE or UNSUBSCRIBE frames to the original subscription. + */ + const val ID = "id" + /** + * The valid values for the ack header are auto, client, or client-individual. + * If the header is not set, it defaults to auto. + * + * For detail information see https://stomp.github.io + */ + const val ACK = "ack" + /** + * The header matching the identifier of the subscription that is receiving the message. + */ + const val SUBSCRIPTION = "subscription" + /** + * The header with a unique identifier for that message. + */ + const val MESSAGE_ID = "message-id" + /** + * The header receipt-id, where the value is the value of the receipt header in the frame which this is a receipt for. + */ + const val RECEIPT_ID = "receipt-id" + } + + val destination: String? + get() = headers[DESTINATION] + + /** + * Get heartBeat in pair format where first value is sendInterval, second value is receiveInterval. + */ + val heartBeat: Pair + get() { + val heartBeat = headers[HEARTBEAT] ?: return 0L to 0L + val (sendInterval, receiveInterval) = heartBeat.split(',') + return sendInterval.toLong() to receiveInterval.toLong() + } + + override fun toString(): String { + return "StompHeader(headers=$headers)" + } +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/models/StompMessage.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/models/StompMessage.kt new file mode 100644 index 00000000..3bfa4369 --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/models/StompMessage.kt @@ -0,0 +1,77 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.okhttp.models + +import com.tinder.scarlet.stomp.okhttp.support.StompHeaderAccessor + +/** + * Stomp message representation with headers, payload and command. + * @see StompCommand + */ +class StompMessage private constructor( + val command: StompCommand, + val payload: ByteArray, + val headers: StompHeader +) { + + class Builder { + + private var payload: ByteArray = ByteArray(0) + private var headers: StompHeaderAccessor = StompHeaderAccessor.of() + + fun withPayload(payload: ByteArray): Builder { + this.payload = payload.copyOf() + return this + } + + fun withPayload(payload: String): Builder { + this.payload = payload.toByteArray() + return this + } + + fun withHeaders(stompHeader: StompHeader): Builder { + this.headers.putAll(stompHeader) + return this + } + + fun create(command: StompCommand): StompMessage { + val createHeader = headers.createHeader() + if (command.isDestinationRequired) check(!createHeader.destination.isNullOrEmpty()) { "Command $command required destination" } + if (!command.isBodyAllowed) check(payload.isEmpty()) { "Command $command doesn't support body" } + return StompMessage( + command = command, + payload = payload, + headers = createHeader + ) + } + } + + override fun equals(other: Any?): Boolean { + if (this === other) return true + if (javaClass != other?.javaClass) return false + + other as StompMessage + + if (command != other.command) return false + if (!payload.contentEquals(other.payload)) return false + if (headers != other.headers) return false + + return true + } + + override fun hashCode(): Int { + var result = command.hashCode() + result = 31 * result + payload.contentHashCode() + result = 31 * result + headers.hashCode() + return result + } + + override fun toString(): String { + return "StompMessage(" + + "command=$command, " + + "payload=${payload.toString(Charsets.UTF_8)}, " + + "headers=$headers" + + ")" + } +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/support/StompHeaderAccessor.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/support/StompHeaderAccessor.kt new file mode 100644 index 00000000..02e3ddae --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/support/StompHeaderAccessor.kt @@ -0,0 +1,129 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.okhttp.support + +import com.tinder.scarlet.stomp.okhttp.models.StompHeader + +/** + * A base for classes providing strongly typed getters and setters as well as + * behavior around specific categories of headers (e.g. STOMP headers). + * Supports creating new headers, modifying existing headers, + * or copying and modifying existing headers. + */ +class StompHeaderAccessor private constructor(headers: Map) { + + companion object { + + /** + * Custom header for the server to notify anything. + */ + private const val STOMP_MESSAGE_HEADER = "message" + + fun of(vararg headers: Pair): StompHeaderAccessor = + StompHeaderAccessor(mapOf(*headers)) + + fun of(headers: Map = emptyMap()): StompHeaderAccessor = + StompHeaderAccessor(headers) + } + + private val mutableHeaders = mutableMapOf().apply { + putAll(headers) + } + + operator fun set(key: String, value: String) { + mutableHeaders[key] = value + } + + fun putAll(headers: Map) { + mutableHeaders.putAll(headers) + } + + fun message(message: String) { + mutableHeaders[STOMP_MESSAGE_HEADER] = message + } + + var contentLength: Int? + get() = try { + mutableHeaders[StompHeader.CONTENT_LENGTH]?.toInt() + } catch (ex: NumberFormatException) { + null + } + set(value) { + mutableHeaders[StompHeader.CONTENT_LENGTH] = value.toString() + } + + var heartBeat: Pair? + get() = mutableHeaders[StompHeader.HEARTBEAT] + ?.split(',') + ?.let { (sendIntervalStr, receiveIntervalStr) -> + val sendInterval = sendIntervalStr.toLongOrNull() ?: 0 + val receiveInterval = receiveIntervalStr.toLongOrNull() ?: 0 + + sendInterval to receiveInterval + } + set(value) { + if (value != null) { + val (sendInterval, receiveInterval) = value + mutableHeaders[StompHeader.HEARTBEAT] = "$sendInterval,$receiveInterval" + } + } + + var subscriptionId: String? + get() = mutableHeaders[StompHeader.ID] + set(value) { + if (value != null) { + mutableHeaders[StompHeader.ID] = value + } + } + + var destination: String? + get() = mutableHeaders[StompHeader.DESTINATION] + set(value) { + if (value != null) { + mutableHeaders[StompHeader.DESTINATION] = value + } + } + + var acceptVersion: String? + get() = mutableHeaders[StompHeader.ACCEPT_VERSION] + set(value) { + if (value != null) { + mutableHeaders[StompHeader.ACCEPT_VERSION] = value + } + } + + var contentType: String? + get() = mutableHeaders[StompHeader.CONTENT_TYPE] + set(value) { + if (value != null) { + mutableHeaders[StompHeader.CONTENT_TYPE] = value + } + } + + var host: String? + get() = mutableHeaders[StompHeader.HOST] + set(value) { + if (value != null) { + mutableHeaders[StompHeader.HOST] = value + } + } + + var login: String? + get() = mutableHeaders[StompHeader.LOGIN] + set(value) { + if (value != null) { + mutableHeaders[StompHeader.LOGIN] = value + } + } + + var passcode: String? + get() = mutableHeaders[StompHeader.PASSCODE] + set(value) { + if (value != null) { + mutableHeaders[StompHeader.PASSCODE] = value + } + } + + fun createHeader() = StompHeader(mutableHeaders) +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/support/StompMessageDecoder.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/support/StompMessageDecoder.kt new file mode 100644 index 00000000..83ea39c5 --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/support/StompMessageDecoder.kt @@ -0,0 +1,194 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.okhttp.support + +import com.tinder.scarlet.stomp.okhttp.models.StompCommand +import com.tinder.scarlet.stomp.okhttp.models.StompMessage +import okio.BufferedSource +import okio.buffer +import okio.source +import java.io.ByteArrayOutputStream + +/** + * An decoder for STOMP frames. + */ +class StompMessageDecoder { + + /** + * Decodes the given byte array into a StompMessage. + * @param array the array to decode + */ + fun decode(array: ByteArray): StompMessage? { + val buffer = array.inputStream().source().buffer() + return decode(buffer) + } + + fun decode(byteBuffer: BufferedSource): StompMessage? { + skipLeadingEol(byteBuffer) + val stompCommand = readCommand(byteBuffer) ?: return null + + return if (stompCommand != StompCommand.HEARTBEAT) { + val headerAccessor = StompHeaderAccessor.of() + + val payload = if (!byteBuffer.exhausted()) { + readHeaders(byteBuffer, headerAccessor) + readPayloadOrNull(byteBuffer, headerAccessor) ?: return null + } else { + ByteArray(0) + } + + StompMessage.Builder() + .withHeaders(headerAccessor.createHeader()) + .withPayload(payload) + .create(stompCommand) + } else { + StompMessage.Builder() + .create(StompCommand.HEARTBEAT) + } + } + + private fun skipLeadingEol(byteBuffer: BufferedSource) { + while (true) { + if (!tryConsumeEndOfLine(byteBuffer)) break + } + } + + private fun readPayloadOrNull( + bufferedSource: BufferedSource, + headerAccessor: StompHeaderAccessor + ): ByteArray? { + val contentLength = headerAccessor.contentLength + return if (contentLength != null && contentLength >= 0) { + readPayloadWithContentLength(bufferedSource, contentLength) + } else { + readPayloadWithoutContentLength(bufferedSource) + } + } + + private fun readPayloadWithContentLength( + bufferedSource: BufferedSource, + contentLength: Int + ): ByteArray? { + if (bufferedSource.exhausted()) return null + + val payload = ByteArray(contentLength) + bufferedSource.read(payload) + + if (bufferedSource.exhausted()) return null + val lastSymbolIsNullOctet = bufferedSource.readUtf8CodePoint() == 0 + check(lastSymbolIsNullOctet) { "Frame must be terminated with a null octet" } + + return payload + } + + private fun readPayloadWithoutContentLength(buffer: BufferedSource): ByteArray? { + val payload = ByteArrayOutputStream(256) + while (!buffer.exhausted()) { + val codePoint = buffer.readUtf8CodePoint() + if (codePoint != 0) { + payload.write(codePoint) + } else { + return payload.toByteArray() + } + } + return null + } + + private fun readHeaders(byteBuffer: BufferedSource, headerAccessor: StompHeaderAccessor) { + while (true) { + val headerStream = ByteArrayOutputStream(256) + var headerComplete = false + + while (!byteBuffer.exhausted()) { + if (tryConsumeEndOfLine(byteBuffer)) { + headerComplete = true + break + } + headerStream.write(byteBuffer.readUtf8CodePoint()) + } + + if (headerStream.size() > 0 && headerComplete) { + val header = headerStream.toByteArray().toString(Charsets.UTF_8) + val colonIndex = header.indexOf(':') + + if (colonIndex > 0) { + val headerName = unescape(header.substring(0, colonIndex)) + val headerValue = unescape(header.substring(colonIndex + 1)) + + headerAccessor[headerName] = headerValue + } else { + check(byteBuffer.exhausted()) { "Illegal header: '$header'. A header must be of the form :[]." } + } + } else { + break + } + } + } + + private fun readCommand(byteBuffer: BufferedSource): StompCommand? { + val command = ByteArrayOutputStream(256) + while (!byteBuffer.exhausted() && !tryConsumeEndOfLine(byteBuffer)) { + command.write(byteBuffer.readUtf8CodePoint()) + } + val commandString = command.toByteArray().toString(Charsets.UTF_8) + return try { + if (commandString.isNotEmpty()) { + StompCommand.valueOf(commandString) + } else { + StompCommand.HEARTBEAT + } + } catch (ex: Exception) { + null + } + } + + /** + * Try to read an EOL incrementing the buffer position if successful. + * @return whether an EOL was consumed + */ + private fun tryConsumeEndOfLine(bufferedSource: BufferedSource): Boolean { + if (bufferedSource.exhausted()) return false + val peekSource = bufferedSource.peek() + + return when (peekSource.readUtf8CodePoint().toChar()) { + '\n' -> { + bufferedSource.skip(1) + true + } + '\r' -> { + val nextChartIsNewLine = peekSource.readUtf8CodePoint().toChar() == '\n' + check(!peekSource.exhausted() && nextChartIsNewLine) { "'\\r' must be followed by '\\n'" } + bufferedSource.skip(2) + true + } + else -> false + } + } + + /** + * See STOMP Spec 1.2: + * ["Value Encoding"](https://stomp.github.io/stomp-specification-1.2.html#Value_Encoding). + */ + private fun unescape(inString: String): String { + val stringBuilder = StringBuilder(inString.length) + var pos = 0 // position in the old string + var index = inString.indexOf('\\') + while (index >= 0) { + stringBuilder.append(inString.substring(pos, index)) + check(index + 1 < inString.length) { "Illegal escape sequence at index $index: $inString" } + + when (inString[index + 1]) { + 'r' -> stringBuilder.append('\r') + 'n' -> stringBuilder.append('\n') + 'c' -> stringBuilder.append(':') + '\\' -> stringBuilder.append('\\') + else -> throw IllegalStateException("Illegal escape sequence at index $index: $inString") // should never happen + } + pos = index + 2 + index = inString.indexOf('\\', pos) + } + stringBuilder.append(inString.substring(pos)) + return stringBuilder.toString() + } +} diff --git a/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/support/StompMessageEncoder.kt b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/support/StompMessageEncoder.kt new file mode 100644 index 00000000..d4e19538 --- /dev/null +++ b/scarlet-protocol-stomp/src/main/java/com/tinder/scarlet/stomp/okhttp/support/StompMessageEncoder.kt @@ -0,0 +1,115 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.okhttp.support + +import com.tinder.scarlet.stomp.okhttp.models.StompCommand +import com.tinder.scarlet.stomp.okhttp.models.StompHeader +import com.tinder.scarlet.stomp.okhttp.models.StompMessage +import java.io.ByteArrayOutputStream +import java.io.DataOutputStream + +/** + * An encoder for STOMP frames. + */ +class StompMessageEncoder { + + companion object { + + private const val LF = '\n'.toInt() + private const val COLON = ':'.toInt() + + private const val HEARTBEAT_PAYLOAD = "\n" + private const val START_HEADERS_SIZE = 64 + } + + /** + * Encodes the given STOMP {@code stompMessage} into a byte array. + * @param stompMessage the message to encode + * @return the encoded message + */ + fun encode(stompMessage: StompMessage): ByteArray { + val arraySize = START_HEADERS_SIZE + stompMessage.payload.size + val arrayOutputStream = ByteArrayOutputStream(arraySize) + + DataOutputStream(arrayOutputStream).use { dataOutputStream -> + + val command = stompMessage.command + if (command != StompCommand.HEARTBEAT) { + writeMessage(dataOutputStream, command, stompMessage) + } else { + dataOutputStream.writeChars(HEARTBEAT_PAYLOAD) + } + } + + return arrayOutputStream.toByteArray() + } + + private fun writeMessage( + dataOutputStream: DataOutputStream, + command: StompCommand, + stompMessage: StompMessage + ) = with(dataOutputStream) { + write(command.toString().toByteArray(Charsets.UTF_8)) + write(LF) + writeHeaders(stompMessage, dataOutputStream) + write(LF) + write(stompMessage.payload) + writeByte(0) + } + + private fun writeHeaders(stompMessage: StompMessage, outputStream: DataOutputStream) { + if (stompMessage.headers.isEmpty()) return + + val command = stompMessage.command + val headers = stompMessage.headers + val shouldEscape = command != StompCommand.CONNECT && command != StompCommand.CONNECTED + + headers.forEach { (key, value) -> + if (command.isBodyAllowed && key == StompHeader.CONTENT_LENGTH) return@forEach + + val headerKey = encode(key, shouldEscape) + val headerValue = encode(value, shouldEscape) + + with(outputStream) { + write(headerKey) + write(COLON) + write(headerValue) + write(LF) + } + } + + if (command.isBodyAllowed) { + val contentLength = stompMessage.payload.size + with(outputStream) { + write(StompHeader.CONTENT_LENGTH.toByteArray(Charsets.UTF_8)) + write(COLON) + write(contentLength.toString().toByteArray(Charsets.UTF_8)) + write(LF) + } + } + } + + private fun encode(input: String, escape: Boolean): ByteArray { + val outputString = if (escape) escape(input) else input + return outputString.toByteArray(Charsets.UTF_8) + } + + /** + * See STOMP Spec 1.2: + * "Value Encoding". + */ + private fun escape(inString: String): String { + val stringBuilder = StringBuilder(inString.length) + inString.forEach { symbol -> + when (symbol) { + '\\' -> stringBuilder.append("\\\\") + ':' -> stringBuilder.append("\\c") + '\n' -> stringBuilder.append("\\n") + '\r' -> stringBuilder.append("\\r") + else -> stringBuilder.append(symbol) + } + } + return stringBuilder.toString() + } +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/test/java/com/tindre/scarlet/stomp/StompIntegrationTest.kt b/scarlet-protocol-stomp/src/test/java/com/tinder/scarlet/stomp/StompIntegrationTest.kt similarity index 75% rename from scarlet-protocol-stomp/src/test/java/com/tindre/scarlet/stomp/StompIntegrationTest.kt rename to scarlet-protocol-stomp/src/test/java/com/tinder/scarlet/stomp/StompIntegrationTest.kt index 9bcaa347..a7488cb5 100644 --- a/scarlet-protocol-stomp/src/test/java/com/tindre/scarlet/stomp/StompIntegrationTest.kt +++ b/scarlet-protocol-stomp/src/test/java/com/tinder/scarlet/stomp/StompIntegrationTest.kt @@ -2,7 +2,7 @@ * © 2018 Match Group, LLC. */ -package com.tindre.scarlet.stomp +package com.tinder.scarlet.stomp import com.tinder.scarlet.ProtocolEvent import com.tinder.scarlet.Stream @@ -10,6 +10,7 @@ import com.tinder.scarlet.testutils.rule.GozirraStompConnection import com.tinder.scarlet.testutils.test import com.tinder.scarlet.ws.Receive import com.tinder.scarlet.ws.Send +import org.apache.activemq.command.ActiveMQDestination import org.apache.activemq.junit.EmbeddedActiveMQBroker import org.apache.activemq.transport.stomp.StompConnection import org.junit.Rule @@ -20,6 +21,8 @@ class StompIntegrationTest { @get:Rule val broker = object : EmbeddedActiveMQBroker() { override fun configure() { + val destination = ActiveMQDestination.createDestination(SERVER_DESTINATION, 0) + brokerService.destinations = arrayOf(destination) brokerService.addConnector(BROKER_URL) } } @@ -31,7 +34,7 @@ class StompIntegrationTest { PORT, LOGIN, PASSWORD, - DESTINATION + CLIENT_DESTINATION ) ) @get:Rule @@ -42,7 +45,7 @@ class StompIntegrationTest { PORT, LOGIN, PASSWORD, - DESTINATION + CLIENT_DESTINATION ) ) @@ -58,17 +61,23 @@ class StompIntegrationTest { connection2.open() LOGGER.info("${queueTextObserver.values}") - queueTextObserver.awaitCount(2) + queueTextObserver.awaitCountAndCheck(2) } @Test fun test2() { val connection1 = StompConnection() - connection1.open(HOST, PORT) - connection1.connect(LOGIN, PASSWORD) + connection1.open( + HOST, + PORT + ) + connection1.connect( + LOGIN, + PASSWORD + ) connection1.begin("tx1") - connection1.send(DESTINATION, "message1", "tx1", null) - connection1.send(DESTINATION, "message2", "tx1", null) + connection1.send(CLIENT_DESTINATION, "message1", "tx1", null) + connection1.send(CLIENT_DESTINATION, "message2", "tx1", null) connection1.commit("tx1") connection1.disconnect() @@ -76,7 +85,7 @@ class StompIntegrationTest { val queueTextObserver = connection2.client.observeText().test() - queueTextObserver.awaitCount(2) + queueTextObserver.awaitCountAndCheck(2) LOGGER.info("${queueTextObserver.values}") } @@ -89,7 +98,8 @@ class StompIntegrationTest { private const val LOGIN = "system" private const val PASSWORD = "manager" private const val BROKER_URL = "stomp://$HOST:$PORT" - private const val DESTINATION = "/queue/test" + private const val SERVER_DESTINATION = "queue://test" + private const val CLIENT_DESTINATION = "/queue/test" interface StompQueueTestService { @Receive diff --git a/scarlet-protocol-stomp/src/test/java/com/tinder/scarlet/stomp/okhttp/OkHttpStompIntegrationTest.kt b/scarlet-protocol-stomp/src/test/java/com/tinder/scarlet/stomp/okhttp/OkHttpStompIntegrationTest.kt new file mode 100644 index 00000000..3bc1b959 --- /dev/null +++ b/scarlet-protocol-stomp/src/test/java/com/tinder/scarlet/stomp/okhttp/OkHttpStompIntegrationTest.kt @@ -0,0 +1,143 @@ +package com.tinder.scarlet.stomp.okhttp + +import com.tinder.scarlet.ProtocolEvent +import com.tinder.scarlet.Stream +import com.tinder.scarlet.testutils.rule.OkHttpStompWebSocketConnection +import com.tinder.scarlet.testutils.test +import com.tinder.scarlet.ws.Receive +import com.tinder.scarlet.ws.Send +import io.reactivex.observers.BaseTestConsumer.TestWaitStrategy.SLEEP_100MS +import org.apache.activemq.command.ActiveMQDestination +import org.apache.activemq.junit.EmbeddedActiveMQBroker +import org.junit.After +import org.junit.Before +import org.junit.Rule +import org.junit.Test +import java.util.logging.Logger + +class OkHttpStompIntegrationTest { + + private var broker: TestEmbeddedActiveMQBroker = TestEmbeddedActiveMQBroker() + + @get:Rule + val firstConnection = OkHttpStompWebSocketConnection.create( + { observeProtocolEvent() }, + OkHttpStompWebSocketConnection.Configuration( + login = LOGIN, + port = PORT, + password = PASSWORD, + host = HOST, + destination = CLIENT_DESTINATION + ) + ) + + @get:Rule + val secondConnection = OkHttpStompWebSocketConnection.create( + { observeProtocolEvent() }, + OkHttpStompWebSocketConnection.Configuration( + login = LOGIN, + port = PORT, + password = PASSWORD, + host = HOST, + destination = CLIENT_DESTINATION + ) + ) + + @Before + fun setUp() { + startServer() + } + + @After + fun tearDown() { + stopServer() + } + + @Test + fun `correct receive and send messages`() { + val queueTextObserver = secondConnection.client.observeText().test() + + firstConnection.open() + secondConnection.open() + + for (index in 0 until 9) { + firstConnection.client.sendText("message $index") + } + firstConnection.clientClosure() + + LOGGER.info("${queueTextObserver.values}") + queueTextObserver.awaitCountAndCheck(9, SLEEP_100MS) + secondConnection.clientClosure() + } + + @Test + fun `reconnect test`() { + val queueTextObserver = secondConnection.client.observeText().test() + + firstConnection.open() + secondConnection.open() + + restartServer() + + firstConnection.clientProtocolEventObserver.awaitCountAndCheck(3) // Open -> Close -> Open + secondConnection.clientProtocolEventObserver.awaitCountAndCheck(3) // Open -> Close -> Open + + for (index in 0 until 9) { + firstConnection.client.sendText("message $index") + } + + firstConnection.clientClosure() + + LOGGER.info("${queueTextObserver.values}") + queueTextObserver.awaitCountAndCheck(9, SLEEP_100MS) + secondConnection.clientClosure() + } + + private fun startServer() { + broker = TestEmbeddedActiveMQBroker() + broker.start() + } + + private fun stopServer() { + broker.stop() + } + + private fun restartServer() { + broker.stop() + broker = TestEmbeddedActiveMQBroker() + broker.start() + } + + private class TestEmbeddedActiveMQBroker : EmbeddedActiveMQBroker() { + + override fun configure() { + val destination = ActiveMQDestination.createDestination(SERVER_DESTINATION, 0) + brokerService.destinations = arrayOf(destination) + brokerService.addConnector(BROKER_URL) + brokerService.isPersistent = false + } + } + + companion object { + private val LOGGER = Logger.getLogger(OkHttpStompIntegrationTest::class.java.name) + + private const val HOST = "localhost" + private const val PORT = 61613 + private const val LOGIN = "system" + private const val PASSWORD = "manager" + private const val BROKER_URL = "ws://$HOST:$PORT" + private const val SERVER_DESTINATION = "topic://queue_test" + private const val CLIENT_DESTINATION = "/topic/test" + + interface StompOkHttpQueueTestService { + @Receive + fun observeProtocolEvent(): Stream + + @Receive + fun observeText(): Stream + + @Send + fun sendText(message: String) + } + } +} diff --git a/scarlet-protocol-stomp/src/test/java/com/tinder/scarlet/stomp/okhttp/client/WebSocketConnectionTest.kt b/scarlet-protocol-stomp/src/test/java/com/tinder/scarlet/stomp/okhttp/client/WebSocketConnectionTest.kt new file mode 100644 index 00000000..87e6bc34 --- /dev/null +++ b/scarlet-protocol-stomp/src/test/java/com/tinder/scarlet/stomp/okhttp/client/WebSocketConnectionTest.kt @@ -0,0 +1,86 @@ +package com.tinder.scarlet.stomp.okhttp.client + +import com.nhaarman.mockito_kotlin.mock +import com.tinder.scarlet.stomp.okhttp.core.Connection +import okhttp3.WebSocket +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test +import java.util.concurrent.Executors +import java.util.concurrent.ScheduledExecutorService +import java.util.concurrent.TimeUnit + +class WebSocketConnectionTest { + + companion object { + private const val TEST_DURATION = 100L + private const val TEST_LONG_DURATION = 10_000L + } + + private lateinit var connection: Connection + private lateinit var scheduledExecutorService: ScheduledExecutorService + private lateinit var webSocket: WebSocket + + @Before + fun setUp() { + webSocket = mock() + scheduledExecutorService = Executors.newSingleThreadScheduledExecutor() + + connection = WebSocketConnection(webSocket, scheduledExecutorService) + } + + @Test(expected = IllegalStateException::class) + fun `cannot set receive callback with 0 duration`() { + connection.onReceiveInactivity(0) {} + } + + @Test(expected = IllegalStateException::class) + fun `cannot set write callback with 0 duration`() { + connection.onWriteInactivity(0) {} + } + + @Test + fun `check that receive callback will be invoked on time`() { + var wasInvoked = false + connection.onReceiveInactivity(TEST_DURATION) { + wasInvoked = true + } + + scheduledExecutorService.awaitTermination(TEST_DURATION * 2, TimeUnit.MILLISECONDS) + assertTrue(wasInvoked) + } + + @Test + fun `check that receive callback won't be invoked if time did't come`() { + var wasInvoked = false + connection.onReceiveInactivity(TEST_LONG_DURATION) { + wasInvoked = true + } + + scheduledExecutorService.awaitTermination(TEST_DURATION * 2, TimeUnit.MILLISECONDS) + assertFalse(wasInvoked) + } + + @Test + fun `check that write callback will be invoked on time`() { + var wasInvoked = false + connection.onWriteInactivity(TEST_DURATION) { + wasInvoked = true + } + + scheduledExecutorService.awaitTermination(TEST_DURATION * 2, TimeUnit.MILLISECONDS) + assertTrue(wasInvoked) + } + + @Test + fun `check that write callback won't be invoked if time did't come`() { + var wasInvoked = false + connection.onWriteInactivity(TEST_LONG_DURATION) { + wasInvoked = true + } + + scheduledExecutorService.awaitTermination(TEST_DURATION * 2, TimeUnit.MILLISECONDS) + assertFalse(wasInvoked) + } +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/test/java/com/tinder/scarlet/stomp/support/StompMessageDecoderTest.kt b/scarlet-protocol-stomp/src/test/java/com/tinder/scarlet/stomp/support/StompMessageDecoderTest.kt new file mode 100644 index 00000000..8e9c74d7 --- /dev/null +++ b/scarlet-protocol-stomp/src/test/java/com/tinder/scarlet/stomp/support/StompMessageDecoderTest.kt @@ -0,0 +1,197 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.support + +import com.tinder.scarlet.stomp.okhttp.models.StompCommand +import com.tinder.scarlet.stomp.okhttp.models.StompHeader +import com.tinder.scarlet.stomp.okhttp.models.StompMessage +import com.tinder.scarlet.stomp.okhttp.support.StompMessageDecoder +import junit.framework.Assert.assertEquals +import junit.framework.Assert.assertNull +import org.junit.Test + +class StompMessageDecoderTest { + + private val decoder = StompMessageDecoder() + + @Test + fun `decode message with CRLF EOls`() { + val message = decode("DISCONNECT\r\n\r\n\u0000") + assertEquals( + StompCommand.DISCONNECT, + message.command + ) + assertEquals(0, message.headers.size) + assertEquals(0, message.payload.size) + } + + @Test + fun `decode message with no headers and no body`() { + val message = decode("DISCONNECT\n\n\u0000") + assertEquals( + StompCommand.DISCONNECT, + message.command + ) + assertEquals(0, message.headers.size) + assertEquals(0, message.payload.size) + } + + @Test + fun `decode message with no body`() { + val accept = "accept-version:1.1\n" + val host = "host:github.org\n" + val message = decode("CONNECT\n$accept$host\n\u0000") + + assertEquals( + StompCommand.CONNECT, + message.command + ) + assertEquals(2, message.headers.size) + assertEquals("1.1", message.headers[StompHeader.ACCEPT_VERSION]) + assertEquals("github.org", message.headers[StompHeader.HOST]) + assertEquals(0, message.payload.size) + } + + @Test + fun `decode message`() { + val message = decode("MESSAGE\ndestination:test\n\nThe body of the message\u0000") + + assertEquals(StompCommand.MESSAGE, message.command) + assertEquals(message.headers.toString(), 1, message.headers.size) + assertEquals("test", message.headers.destination) + + val bodyText = message.payload.toString(Charsets.UTF_8) + assertEquals("The body of the message", bodyText) + } + + @Test + fun `decode message with content length`() { + val message = + decode("MESSAGE\ndestination:test\ncontent-length:23\n\nThe body of the message\u0000") + + assertEquals(StompCommand.MESSAGE, message.command) + assertEquals(2, message.headers.size) + assertEquals(23, message.headers[StompHeader.CONTENT_LENGTH]?.toInt()) + + val bodyText = message.payload.toString(Charsets.UTF_8) + assertEquals("The body of the message", bodyText) + } + + @Test + fun `decode frame with invalid content length`() { + val message = + decode("MESSAGE\ndestination:test\ncontent-length:-1\n\nThe body of the message\u0000") + + assertEquals(StompCommand.MESSAGE, message.command) + assertEquals(2, message.headers.size) + assertEquals(-1, message.headers[StompHeader.CONTENT_LENGTH]?.toInt()) + + val bodyText = message.payload.toString(Charsets.UTF_8) + assertEquals("The body of the message", bodyText) + } + + @Test + fun `decode frame with content length zero`() { + val message = decode("MESSAGE\ndestination:test\ncontent-length:0\n\n\u0000") + + assertEquals(StompCommand.MESSAGE, message.command) + assertEquals(2, message.headers.size) + assertEquals(0, message.headers[StompHeader.CONTENT_LENGTH]?.toInt()) + + val bodyText = message.payload.toString(Charsets.UTF_8) + assertEquals("", bodyText) + } + + @Test + fun `decode frame with null octects in the body`() { + val message = + decode("MESSAGE\ndestination:test\ncontent-length:23\n\nThe b\u0000dy \u0000f the message\u0000") + + assertEquals(StompCommand.MESSAGE, message.command) + assertEquals(2, message.headers.size) + assertEquals(23, message.headers[StompHeader.CONTENT_LENGTH]?.toInt()) + + val bodyText = message.payload.toString(Charsets.UTF_8) + assertEquals("The b\u0000dy \u0000f the message", bodyText) + } + + @Test + fun `decode frame with escaped headers`() { + val message = + decode("CONNECTED\na\\c\\r\\n\\\\b:alpha\\cbravo\\r\\n\\\\\n\n\u0000") + assertEquals( + StompCommand.CONNECTED, + message.command + ) + assertEquals(1, message.headers.size) + assertEquals("alpha:bravo\r\n\\", message.headers["a:\r\n\\b"]) + } + + @Test(expected = IllegalStateException::class) + fun `decode frame body not allowed`() { + decode("CONNECTED\naccept-version:1.2\n\nThe body of the message\u0000") + } + + @Test + fun `decode frame with header with empty value`() { + val accept = "accept-version:1.1\n" + val valuelessKey = "key:\n" + val message = decode("CONNECTED\n$accept$valuelessKey\n\u0000") + + assertEquals( + StompCommand.CONNECTED, + message.command + ) + assertEquals(2, message.headers.size) + assertEquals("1.1", message.headers["accept-version"]) + assertEquals("", message.headers["key"]) + assertEquals(0, message.payload.size) + } + + @Test + fun decodeFrameWithIncompleteCommand() { + assertIncompleteDecode("MESSAG") + } + + @Test + fun `decode frame with incomplete header`() { + assertIncompleteDecode("SEND\ndestination") + assertIncompleteDecode("SEND\ndestination:") + assertIncompleteDecode("SEND\ndestination:test") + } + + @Test + fun `decode frame without null octet terminator`() { + assertIncompleteDecode("SEND\ndestination:test\n") + assertIncompleteDecode("SEND\ndestination:test\n\n") + assertIncompleteDecode("SEND\ndestination:test\n\nThe body") + } + + @Test + fun `decode frame with insufficient content`() { + assertIncompleteDecode("SEND\ncontent-length:23\n\nThe body of the mess") + } + + @Test + fun decodeHeartbeat() { + val frame = "\n" + val array = frame.toByteArray() + val message = decoder.decode(array) + + assertEquals( + StompCommand.HEARTBEAT, + message?.command + ) + } + + private fun assertIncompleteDecode(frame: String) { + val message = decoder.decode(frame.toByteArray(Charsets.UTF_8)) + assertNull(message) + } + + private fun decode(stompFrame: String): StompMessage { + val array = stompFrame.toByteArray() + return array.let(decoder::decode)!! + } +} \ No newline at end of file diff --git a/scarlet-protocol-stomp/src/test/java/com/tinder/scarlet/stomp/support/StompMessageEncoderTest.kt b/scarlet-protocol-stomp/src/test/java/com/tinder/scarlet/stomp/support/StompMessageEncoderTest.kt new file mode 100644 index 00000000..6e79a2e9 --- /dev/null +++ b/scarlet-protocol-stomp/src/test/java/com/tinder/scarlet/stomp/support/StompMessageEncoderTest.kt @@ -0,0 +1,96 @@ +/* + * © 2018 Match Group, LLC. + */ +package com.tinder.scarlet.stomp.support + +import com.tinder.scarlet.stomp.okhttp.models.StompCommand +import com.tinder.scarlet.stomp.okhttp.models.StompMessage +import com.tinder.scarlet.stomp.okhttp.support.StompHeaderAccessor +import com.tinder.scarlet.stomp.okhttp.support.StompMessageEncoder +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Test + +class StompMessageEncoderTest { + + private val encoder = StompMessageEncoder() + + @Test + fun `encode frame with no headers and nobBody`() { + val frame = StompMessage.Builder() + .withPayload(ByteArray(0)) + .create(StompCommand.DISCONNECT) + + assertEquals("DISCONNECT\n\n\u0000", String(encoder.encode(frame))) + } + + @Test + fun `encode frame with headers`() { + val headers = StompHeaderAccessor.of().apply { + acceptVersion = "1.2" + host = "github.org" + }.createHeader() + + val frame = StompMessage.Builder() + .withPayload(ByteArray(0)) + .withHeaders(headers) + .create(StompCommand.CONNECT) + + val frameString = String(encoder.encode(frame)) + assertTrue( + "CONNECT\naccept-version:1.2\nhost:github.org\n\n\u0000" == frameString || + "CONNECT\nhost:github.org\naccept-version:1.2\n\n\u0000" == frameString + ) + } + + @Test + fun `encode frame with headers that should be escaped`() { + val headers = StompHeaderAccessor.of("a:\r\n\\b" to "alpha:bravo\r\n\\") + .createHeader() + + val frame = StompMessage.Builder() + .withPayload(ByteArray(0)) + .withHeaders(headers) + .create(StompCommand.DISCONNECT) + + assertEquals( + "DISCONNECT\na\\c\\r\\n\\\\b:alpha\\cbravo\\r\\n\\\\\n\n\u0000", + String(encoder.encode(frame)) + ) + } + + @Test + fun `encode frame with headers body`() { + val headers = StompHeaderAccessor.of(("a" to "alpha")).apply { + destination = "destination" + }.createHeader() + + val frame = StompMessage.Builder() + .withPayload("Message body") + .withHeaders(headers) + .create(StompCommand.SEND) + + assertEquals( + "SEND\na:alpha\ndestination:destination\ncontent-length:12\n\nMessage body\u0000", + String(encoder.encode(frame)) + ) + } + + @Test + fun `encode frame with content length present`() { + val headers = StompHeaderAccessor.of().apply { + contentLength = 22 + destination = "destination" + }.createHeader() + + val frame = StompMessage.Builder() + .withPayload("Message body") + .withHeaders(headers) + .create(StompCommand.SEND) + + assertEquals( + "SEND\ndestination:destination\ncontent-length:12\n\nMessage body\u0000", + String(encoder.encode(frame)) + ) + } +} \ No newline at end of file diff --git a/scarlet-stream-adapter-coroutines/src/test/java/com/tinder/scarlet/streamadapter/coroutines/FlowStreamAdapterTest.kt b/scarlet-stream-adapter-coroutines/src/test/java/com/tinder/scarlet/streamadapter/coroutines/FlowStreamAdapterTest.kt index 7e26b4c6..6f2c144d 100644 --- a/scarlet-stream-adapter-coroutines/src/test/java/com/tinder/scarlet/streamadapter/coroutines/FlowStreamAdapterTest.kt +++ b/scarlet-stream-adapter-coroutines/src/test/java/com/tinder/scarlet/streamadapter/coroutines/FlowStreamAdapterTest.kt @@ -47,7 +47,6 @@ class FlowStreamAdapterTest { testTextFlow.collect { receivedTextItems.add(it) } - } val job2 = launch { testBytesFlow.collect { diff --git a/scarlet-test-utils/src/main/java/com/tinder/scarlet/testutils/TestStreamObserver.kt b/scarlet-test-utils/src/main/java/com/tinder/scarlet/testutils/TestStreamObserver.kt index 71e14610..05efa8c4 100644 --- a/scarlet-test-utils/src/main/java/com/tinder/scarlet/testutils/TestStreamObserver.kt +++ b/scarlet-test-utils/src/main/java/com/tinder/scarlet/testutils/TestStreamObserver.kt @@ -5,6 +5,7 @@ package com.tinder.scarlet.testutils import com.tinder.scarlet.Stream +import io.reactivex.observers.BaseTestConsumer.TestWaitStrategy import io.reactivex.processors.PublishProcessor import org.assertj.core.api.Assertions.assertThat @@ -25,14 +26,14 @@ class TestStreamObserver(stream: Stream) { val completions: Long get() = testSubscriber.completions() - fun awaitCount(exactly: Int) { + fun awaitCountAndCheck(exactly: Int, waitStrategy: Runnable = TestWaitStrategy.SLEEP_10MS) { testSubscriber.assertNoErrors() - testSubscriber.awaitCount(exactly) + testSubscriber.awaitCount(exactly, waitStrategy) assertThat(values).describedAs("values: $values").hasSize(exactly) } fun awaitValues(vararg valueAsserts: ValueAssert) { - awaitCount(valueAsserts.size) + awaitCountAndCheck(valueAsserts.size) testSubscriber.assertNoErrors() valueAsserts.zip(values).forEachIndexed { index, (valueAssert, value) -> try { diff --git a/scarlet-test-utils/src/main/java/com/tinder/scarlet/testutils/rule/OkHttpStompWebSocketConnection.kt b/scarlet-test-utils/src/main/java/com/tinder/scarlet/testutils/rule/OkHttpStompWebSocketConnection.kt new file mode 100644 index 00000000..24465c81 --- /dev/null +++ b/scarlet-test-utils/src/main/java/com/tinder/scarlet/testutils/rule/OkHttpStompWebSocketConnection.kt @@ -0,0 +1,163 @@ +/* + * © 2018 Match Group, LLC. + */ + +package com.tinder.scarlet.testutils.rule + +import com.tinder.scarlet.LifecycleState +import com.tinder.scarlet.MessageAdapter +import com.tinder.scarlet.ProtocolEvent +import com.tinder.scarlet.Scarlet +import com.tinder.scarlet.Stream +import com.tinder.scarlet.StreamAdapter +import com.tinder.scarlet.lifecycle.LifecycleRegistry +import com.tinder.scarlet.stomp.okhttp.client.OkHttpStompClient +import com.tinder.scarlet.stomp.okhttp.client.OkHttpStompDestination +import com.tinder.scarlet.stomp.okhttp.client.OkHttpStompMainChannel +import com.tinder.scarlet.testutils.TestStreamObserver +import com.tinder.scarlet.testutils.any +import com.tinder.scarlet.testutils.test +import okhttp3.OkHttpClient +import okhttp3.Request +import org.junit.rules.ExternalResource +import org.junit.rules.TestRule +import org.junit.runner.Description +import org.junit.runners.model.Statement +import java.util.concurrent.TimeUnit +import java.util.logging.Level +import java.util.logging.Logger + +class OkHttpStompWebSocketConnection( + private val clazz: Class, + private val observeProtocolEvent: SERVICE.() -> Stream, + private val clientConfiguration: Configuration +) : TestRule { + + val client: SERVICE + get() = clientAndServer.client + + val clientProtocolEventObserver: TestStreamObserver + get() = clientAndServer.clientProtocolEventObserver + + private val clientLifecycleRegistry = LifecycleRegistry() + + private val clientAndServer = ClientAndServer() + + override fun apply(base: Statement, description: Description): Statement { + return clientAndServer + .apply(base, description) + } + + fun open() { + clientLifecycleRegistry.onNext(LifecycleState.Started) + blockUntilConnectionIsEstablish() + } + + fun clientClosure() { + clientLifecycleRegistry.onNext(LifecycleState.Stopped) + } + + fun clientTerminate() { + clientLifecycleRegistry.onNext(LifecycleState.Completed) + } + + private fun blockUntilConnectionIsEstablish() { + clientProtocolEventObserver.awaitValues( + any() + ) + } + + private inner class ClientAndServer : ExternalResource() { + lateinit var client: SERVICE + lateinit var clientProtocolEventObserver: TestStreamObserver + + override fun before() { + client = createClient() + clientProtocolEventObserver = client.observeProtocolEvent().test() + client.observeProtocolEvent().start(object : Stream.Observer { + override fun onNext(data: ProtocolEvent) { + LOGGER.info("$this: client stomp event: $data") + } + + override fun onError(throwable: Throwable) { + LOGGER.log( + Level.WARNING, + "$this: client stomp error", + throwable + ) + } + + override fun onComplete() { + LOGGER.info("client stomp completed") + } + }) + } + + override fun after() { + clientLifecycleRegistry.onNext(LifecycleState.Completed) + } + + private fun createClient(): SERVICE { + val protocol = OkHttpStompClient( + configuration = OkHttpStompMainChannel.Configuration( + host = "ws://${clientConfiguration.host}:${clientConfiguration.port}" + ), + okHttpClient = createOkHttpClient(), + requestFactory = { + OkHttpStompClient.ClientOpenRequest( + passcode = clientConfiguration.password, + login = clientConfiguration.login, + okHttpRequest = Request.Builder().url( + "http://${clientConfiguration.host}:${clientConfiguration.port}" + ).build() + ) + } + ) + val configuration = Scarlet.Configuration( + lifecycle = clientLifecycleRegistry, + messageAdapterFactories = clientConfiguration.messageAdapterFactories, + streamAdapterFactories = clientConfiguration.streamAdapterFactories, + debug = true + ) + val mainScarlet = Scarlet(protocol, configuration) + return mainScarlet + .child( + OkHttpStompDestination(clientConfiguration.destination), + configuration + ) + .create(clazz) + } + + private fun createOkHttpClient(): OkHttpClient { + return OkHttpClient.Builder() + .writeTimeout(500, TimeUnit.MILLISECONDS) + .readTimeout(500, TimeUnit.MILLISECONDS) + .build() + } + } + + data class Configuration( + val host: String, + val port: Int, + val login: String, + val password: String, + val destination: String, + val messageAdapterFactories: List = emptyList(), + val streamAdapterFactories: List = emptyList() + ) + + companion object { + private val LOGGER = Logger.getLogger(OkHttpStompWebSocketConnection::class.java.name) + + inline fun create( + noinline observeProtocolEvent: SERVICE.() -> Stream, + clientConfiguration: Configuration + ): OkHttpStompWebSocketConnection { + return OkHttpStompWebSocketConnection( + SERVICE::class.java, + observeProtocolEvent, + clientConfiguration + ) + } + } +} \ No newline at end of file