Rewrite streaming API to use websocket

USAGE.md

@@ -208,8 +208,6 @@ public class GetRawJson {
 
 ## Streaming API
 
-v1.0.0 or later
-
 __Kotlin__
 
 ```kotlin
@@ -218,47 +216,26 @@ val client: MastodonClient = MastodonClient.Builder(instanceHostname)
   .useStreamingApi()
   .build()
 
-val handler = object : Handler {
-  override fun onStatus(status: Status) {
-    println(status.content)
-  }
-  override fun onNotification(notification: Notification) {/* no op */}
-  override fun onDelete(id: String) {/* no op */}
-}
-
-try {
-  val shutdownable = client.streaming.localPublic(handler)
-  Thread.sleep(10000L)
-  shutdownable.shutdown()
-} catch(e: BigBoneRequestException) {
-  e.printStackTrace()
+client.streaming.federatedPublic(
+    onlyMedia = false,
+    callback = { event: WebSocketEvent ->
+        println(event)
+    }
+).use {
+    Thread.sleep(15_000L)
 }
 ```
 
 __Java__
 
 ```java
-MastodonClient client = new MastodonClient.Builder(instanceHostname)
+final MastodonClient client = new MastodonClient.Builder(instanceHostname)
         .accessToken(accessToken)
         .useStreamingApi()
         .build();
-Handler handler = new Handler() {
-    @Override
-    public void onStatus(@NotNull Status status) {
-        System.out.println(status.getContent());
-    }
 
-    @Override
-    public void onNotification(@NotNull Notification notification) {/* no op */}
-    @Override
-    public void onDelete(String id) {/* no op */}
-};
-
-try {
-    Shutdownable shutdownable = client.streaming().localPublic(handler);
-    Thread.sleep(10000L);
-    shutdownable.shutdown();
-} catch (Exception e) {
-    e.printStackTrace();
+// Start federated timeline streaming and stop after 20 seconds
+try (Closeable ignored = client.streaming().federatedPublic(false, System.out::println)) {
+        Thread.sleep(20_000L);
 }
 ```

bigbone-rx/src/main/kotlin/social/bigbone/rx/RxStreamingMethods.kt

@@ -1,62 +1,122 @@
 package social.bigbone.rx
 
 import io.reactivex.rxjava3.core.BackpressureStrategy
+import io.reactivex.rxjava3.core.Completable
 import io.reactivex.rxjava3.core.Flowable
 import social.bigbone.MastodonClient
-import social.bigbone.api.Handler
-import social.bigbone.api.Shutdownable
-import social.bigbone.api.entity.Notification
-import social.bigbone.api.entity.Status
+import social.bigbone.api.WebSocketCallback
+import social.bigbone.api.entity.streaming.MastodonApiEvent.GenericMessage
+import social.bigbone.api.entity.streaming.MastodonApiEvent.StreamEvent
+import social.bigbone.api.entity.streaming.TechnicalEvent.Closed
+import social.bigbone.api.entity.streaming.TechnicalEvent.Closing
+import social.bigbone.api.entity.streaming.TechnicalEvent.Failure
+import social.bigbone.api.entity.streaming.TechnicalEvent.Open
+import social.bigbone.api.entity.streaming.WebSocketEvent
 import social.bigbone.api.method.StreamingMethods
+import java.io.Closeable
 
 /**
  * Reactive implementation of [StreamingMethods].
  * Allows access to API methods with endpoints having an "api/vX/streaming" prefix.
- * @see <a href="https://docs.joinmastodon.org/methods/streaming/">Mastodon streaming API methods</a>
+ * @see <a href="https://docs.joinmastodon.org/methods/streaming/#streams">Mastodon streaming API methods</a>
  */
 class RxStreamingMethods(client: MastodonClient) {
+
     private val streamingMethods = StreamingMethods(client)
 
-    private fun stream(f: (Handler) -> Shutdownable): Flowable<Status> {
-        return Flowable.create<Status>({ emitter ->
-            val shutdownable = f(object : Handler {
-                override fun onStatus(status: Status) {
-                    emitter.onNext(status)
-                }
+    /**
+     * Verify that the streaming service is alive before connecting to it.
+     *
+     * @return Completable that will complete if the server is “healthy” or emit an error via onError otherwise.
+     * @see <a href="https://docs.joinmastodon.org/methods/streaming/#health">streaming/#health API documentation</a>
+     */
+    fun checkServerHealth(): Completable = Completable.fromAction {
+        streamingMethods.checkServerHealth()
+    }
 
-                override fun onNotification(notification: Notification) {
-                    // no op
-                }
+    /**
+     * Stream all public posts known to this server. Analogous to the federated timeline.
+     *
+     * @param onlyMedia Filter for media attachments. Analogous to the federated timeline with “only media” enabled.
+     */
+    fun federatedPublic(onlyMedia: Boolean): Flowable<WebSocketEvent> = stream {
+        streamingMethods.federatedPublic(onlyMedia, it)
+    }
 
-                override fun onDelete(id: String) {
-                    // no op
-                }
-            })
-            emitter.setCancellable {
-                shutdownable.shutdown()
-            }
-        }, BackpressureStrategy.LATEST)
+    /**
+     * Stream all public posts originating from this server. Analogous to the local timeline.
+     *
+     * @param onlyMedia Filter for media attachments. Analogous to the local timeline with “only media” enabled.
+     */
+    fun localPublic(onlyMedia: Boolean): Flowable<WebSocketEvent> = stream {
+        streamingMethods.localPublic(onlyMedia, it)
     }
 
-    private fun statusStream(f: (Handler) -> Shutdownable): Flowable<Status> {
-        return stream { handler ->
-            f(handler)
-        }
+    /**
+     * Stream all public posts originating from other servers.
+     *
+     * @param onlyMedia Filter for media attachments.
+     */
+    fun remotePublic(onlyMedia: Boolean): Flowable<WebSocketEvent> = stream {
+        streamingMethods.remotePublic(onlyMedia, it)
     }
 
-    private fun tagStream(tag: String, f: (String, Handler) -> Shutdownable): Flowable<Status> {
-        return stream { handler ->
-            f(tag, handler)
-        }
+    /**
+     * Stream all public posts using the hashtag [tagName].
+     *
+     * @param tagName Hashtag the public posts you want to stream should have.
+     * @param onlyFromThisServer Filter for public posts originating from this server.
+     */
+    fun hashtag(
+        tagName: String,
+        onlyFromThisServer: Boolean
+    ): Flowable<WebSocketEvent> = stream { callback ->
+        streamingMethods.hashtag(
+            tagName = tagName,
+            onlyFromThisServer = onlyFromThisServer,
+            callback = callback
+        )
     }
 
-    fun localPublic(): Flowable<Status> = statusStream(streamingMethods::localPublic)
+    /**
+     * Stream all events related to the current user, such as home feed updates and notifications.
+     */
+    fun user(): Flowable<WebSocketEvent> = stream(streamingMethods::user)
 
-    fun federatedPublic(): Flowable<Status> = statusStream(streamingMethods::federatedPublic)
+    /**
+     * Stream all notifications for the current user.
+     */
+    fun userNotifications(): Flowable<WebSocketEvent> = stream(streamingMethods::userNotifications)
 
-    fun localHashtag(tag: String): Flowable<Status> = tagStream(tag, streamingMethods::localHashtag)
+    /**
+     * Stream updates to the list with [listId].
+     *
+     * @param listId List you want to receive updates for.
+     */
+    fun list(listId: String): Flowable<WebSocketEvent> = stream { callback ->
+        streamingMethods.list(
+            listId = listId,
+            callback = callback
+        )
+    }
 
-    fun federatedHashtag(tag: String): Flowable<Status> = tagStream(tag, streamingMethods::federatedHashtag)
+    /**
+     * Stream all updates to direct conversations.
+     */
+    fun directConversations(): Flowable<WebSocketEvent> = stream(streamingMethods::directConversations)
 
-    // TODO user streaming
+    private fun stream(streamMethod: (WebSocketCallback) -> Closeable): Flowable<WebSocketEvent> =
+        Flowable.create({ emitter ->
+            val closeable = streamMethod { webSocketEvent ->
+                when (webSocketEvent) {
+                    is Closed -> emitter.onComplete()
+                    is Failure -> emitter.tryOnError(webSocketEvent.error)
+                    Open,
+                    is Closing,
+                    is StreamEvent,
+                    is GenericMessage -> emitter.onNext(webSocketEvent)
+                }
+            }
+            emitter.setCancellable { closeable.close() }
+        }, BackpressureStrategy.BUFFER)
 }

bigbone-rx/src/test/kotlin/social/bigbone/rx/RxStreamingMethodsTest.kt

@@ -0,0 +1,99 @@
+package social.bigbone.rx
+
+import io.mockk.mockk
+import io.reactivex.rxjava3.schedulers.TestScheduler
+import io.reactivex.rxjava3.subscribers.TestSubscriber
+import okio.IOException
+import org.junit.jupiter.api.Test
+import social.bigbone.api.entity.streaming.MastodonApiEvent
+import social.bigbone.api.entity.streaming.ParsedStreamEvent
+import social.bigbone.api.entity.streaming.StreamType
+import social.bigbone.api.entity.streaming.TechnicalEvent
+import social.bigbone.api.entity.streaming.WebSocketEvent
+import social.bigbone.rx.testtool.MockClient
+
+class RxStreamingMethodsTest {
+
+    private val testScheduler = TestScheduler()
+
+    @Test
+    fun `Given client returning OK, when checking server health, then complete without errors`() {
+        val client = MockClient.mockClearText(clearTextResponse = "OK")
+        val streamingMethods = RxStreamingMethods(client)
+
+        val serverHealth = streamingMethods.checkServerHealth().test()
+
+        with(serverHealth) {
+            assertComplete()
+            assertNoErrors()
+        }
+    }
+
+    @Test
+    fun `Given websocket with 6 events lined up, when streaming federated public timeline, then expect emissions and no errors`() {
+        val mockedEvents: List<WebSocketEvent> = listOf(
+            TechnicalEvent.Open,
+            MastodonApiEvent.StreamEvent(
+                ParsedStreamEvent.FiltersChanged,
+                listOf(StreamType.PUBLIC)
+            ),
+            MastodonApiEvent.StreamEvent(
+                ParsedStreamEvent.StatusCreated(mockk()),
+                listOf(StreamType.PUBLIC)
+            ),
+            MastodonApiEvent.StreamEvent(
+                ParsedStreamEvent.StatusDeleted(deletedStatusId = "12345"),
+                listOf(StreamType.PUBLIC)
+            ),
+            MastodonApiEvent.StreamEvent(
+                ParsedStreamEvent.AnnouncementDeleted(deletedAnnouncementId = "54321"),
+                listOf(StreamType.PUBLIC)
+            ),
+            MastodonApiEvent.StreamEvent(
+                ParsedStreamEvent.StatusCreated(createdStatus = mockk()),
+                listOf(StreamType.PUBLIC)
+            )
+        )
+        val client = MockClient.mockWebSocket(events = mockedEvents)
+        val streamingMethods = RxStreamingMethods(client)
+
+        val testSubscriber: TestSubscriber<WebSocketEvent> = streamingMethods
+            .federatedPublic(onlyMedia = false)
+            .subscribeOn(testScheduler)
+            .observeOn(testScheduler)
+            .test()
+        testScheduler.triggerActions()
+
+        with(testSubscriber) {
+            assertValueCount(mockedEvents.size)
+            assertValueSequence(mockedEvents)
+            assertNotComplete()
+            assertNoErrors()
+            cancel()
+        }
+    }
+
+    @Test
+    fun `Given websocket with failure, when streaming federated public timeline, then expect error is propagated`() {
+        val expectedError = IOException("Expected")
+        val mockedEvents: List<WebSocketEvent> = listOf(
+            TechnicalEvent.Open,
+            TechnicalEvent.Failure(expectedError)
+        )
+        val client = MockClient.mockWebSocket(events = mockedEvents)
+        val streamingMethods = RxStreamingMethods(client)
+
+        val testSubscriber: TestSubscriber<WebSocketEvent> = streamingMethods
+            .federatedPublic(onlyMedia = false)
+            .subscribeOn(testScheduler)
+            .observeOn(testScheduler)
+            .test()
+        testScheduler.triggerActions()
+
+        with(testSubscriber) {
+            assertNotComplete()
+            assertError(expectedError)
+            cancel()
+        }
+    }
+}

bigbone-rx/src/test/kotlin/social/bigbone/rx/testtool/MockClient.kt

@@ -9,13 +9,57 @@ import okhttp3.RequestBody
 import okhttp3.Response
 import okhttp3.ResponseBody
 import okhttp3.ResponseBody.Companion.toResponseBody
+import okhttp3.WebSocket
 import social.bigbone.MastodonClient
 import social.bigbone.Parameters
+import social.bigbone.api.WebSocketCallback
+import social.bigbone.api.entity.streaming.WebSocketEvent
 import social.bigbone.api.exception.BigBoneRequestException
 import java.net.SocketTimeoutException
 
 object MockClient {
 
+    /**
+     * Mocks a [MastodonClient] for functions testing the websocket streaming APIs.
+     *
+     * @param events [WebSocketEvent]s that should be lined up to be returned by the [WebSocketCallback]
+     */
+    fun mockWebSocket(events: Collection<WebSocketEvent>): MastodonClient {
+        val webSocket = mockk<WebSocket> {
+            every { close(any<Int>(), any<String>()) } returns true
+        }
+        return mockk<MastodonClient> {
+            every { stream(any<Parameters>(), any<WebSocketCallback>()) } answers {
+                events.forEach { event: WebSocketEvent -> secondArg<WebSocketCallback>().onEvent(event) }
+                webSocket
+            }
+        }
+    }
+
+    fun mockClearText(
+        clearTextResponse: String,
+        requestUrl: String = "https://example.com"
+    ): MastodonClient {
+        val response: Response = Response.Builder()
+            .code(200)
+            .message("OK")
+            .request(Request.Builder().url(requestUrl).build())
+            .protocol(Protocol.HTTP_1_1)
+            .body(clearTextResponse.toResponseBody())
+            .build()
+
+        return mockk {
+            every { get(any<String>(), any()) } returns response
+            every {
+                this@mockk["performAction"](
+                    any<String>(),
+                    any<MastodonClient.Method>(),
+                    any<Parameters>()
+                )
+            } returns response
+        }
+    }
+
     fun mock(
         jsonName: String,
         maxId: String? = null,