Cleanup and Documentation

GitOrigin-RevId: 1115466d02e3a5e97875020c7056b0b4420fb8d1

Author: jclyne

docs/actions.md

@@ -187,8 +187,10 @@ class GreeterActionModule : KAbstractModule() {
 }
 ```
 
-Misk also supports `rpcCallStyle = "suspending"` for suspending gRPC actions.This is the preferred way to generate server 
-actions if you intend on using coroutines to implement the business logic of your action. See [coroutines](coroutines.md) for more information.
+Misk also supports `rpcCallStyle = "suspending"` for suspending gRPC actions. This is the way to generate server 
+actions if you intend to use coroutines to implement the business logic of your action. The generated interface will 
+then expect you to implement a suspending function instead of a regular blocking function for your action's handler method. 
+See [coroutines](coroutines.md) for more information.
 
 ```kotlin
 wire {
@@ -208,7 +210,8 @@ wire {
 }
 ```
 
-The above will generate a similar action class, but with a suspending action function
+The above Wire Gradle plugin configuration will have a suspending function in the generated interface instead of a 
+regular function. Your implementing class will then look like this:
 
 ```kotlin
 @Singleton

docs/coroutines.md

@@ -1,4 +1,4 @@
-Misk Coroutines Rules
+# Misk Coroutines Rules
 
 Coroutines are cooperative concurrency. If you don't cooperate (ie. suspend regularly), things will not be as efficient 
 and may potentially deadlock.  
@@ -32,7 +32,7 @@ Example
 }```
 
 
-When an action is declared with the `suspend` modifier, it'll be called with a `Dispatcher` that has a single backing
+When an action is declared with the `suspend` modifier, it will be called with a `Dispatcher` that has a single backing
 thread (`runBlocking`). This thread is part of the Jetty Thread Pool and allocated to this specific request, therefore
 it is safe to make blocking calls on. This will also take care of request scoped features that are thread local, 
 such as ActionScoped values, MDC, tracing, etc.
@@ -44,6 +44,10 @@ Follow structured concurrency best practices, including:
  - All coroutines should be in child scopes for the incoming request scope.
  - Don't use `GlobalScope` or create a new `CoroutineScope()` object.
 
+Misk is gradually adding experimental support for Kotlin coroutines. 
+Functionality may be partially implemented or buggy in the short term as support is added to core libraries 
+and implementations. Please report any issues encountered.
+
 
 
 

misk/src/main/kotlin/misk/grpc/GrpcFeatureBinding.kt

@@ -107,9 +107,11 @@ internal class GrpcFeatureBinding(
     webConfig: WebConfig
   ) : FeatureBinding.Factory {
 
-    // This dispatcher is sized to the jetty thread pool size to make sure that
-    // no requests that are currently scheduled on a jetty thread are ever blocked
-    // from reading a streaming request
+    /**
+     * This dispatcher is sized to the jetty thread pool size to make sure that
+     * no requests that are currently scheduled on a jetty thread are ever blocked
+     * from reading a streaming request.
+     */
     private val grpcMessageSourceChannelDispatcher =
       Dispatchers.IO.limitedParallelism(
         parallelism = webConfig.jetty_max_thread_pool_size,
@@ -147,7 +149,7 @@ internal class GrpcFeatureBinding(
         @Suppress("UNCHECKED_CAST") // Assume it's a proto type.
         ProtoAdapter.get(wireAnnotation.responseAdapter) as ProtoAdapter<Any>
       }
-      val isSuspending = action.function.isSuspend
+      val isSuspend = action.function.isSuspend
 
       return if (streamingRequestType != null) {
         @Suppress("UNCHECKED_CAST") // Assume it's a proto type.
@@ -156,7 +158,7 @@ internal class GrpcFeatureBinding(
           responseAdapter = responseAdapter,
           streamingRequest = true,
           streamingResponse = streamingResponse,
-          isSuspend = isSuspending,
+          isSuspend = isSuspend,
           grpcMessageSourceChannelContext = grpcMessageSourceChannelDispatcher,
         )
       } else {
@@ -166,7 +168,7 @@ internal class GrpcFeatureBinding(
           responseAdapter = responseAdapter,
           streamingRequest = false,
           streamingResponse = streamingResponse,
-          isSuspend = isSuspending,
+          isSuspend = isSuspend,
           grpcMessageSourceChannelContext = grpcMessageSourceChannelDispatcher,
         )
       }

misk/src/main/kotlin/misk/logging/DynamicMdcContext.kt

@@ -36,7 +36,7 @@ internal class DynamicMdcContext(
   }
 
   override fun restoreThreadContext(context: CoroutineContext, oldState: MDCContext?) {
-    //NOOP
+    clear()
   }
 
   override fun copyForChild(): CopyableThreadContextElement<MDCContext?> {

misk/src/main/kotlin/misk/web/actions/WebActions.kt

@@ -35,22 +35,26 @@ internal fun WebAction.asChain(
         argsMap[param] = arg
       }
 
-      return if (function.isSuspend) {
+      return if (!function.isSuspend) {
+        function.callBy(argsMap)
+      } else {
         // Handle suspending invocation, this includes building out the context to propagate MDC
         // and action scope.
-        val context = EmptyCoroutineContext +
-          DynamicMdcContext() +
-          if (scope.inScope()) scope.asContextElement() else EmptyCoroutineContext
+        val context = DynamicMdcContext() +
+          if (scope.inScope()) {
+            scope.asContextElement()
+          } else {
+            EmptyCoroutineContext
+          }
 
         runBlocking(context) {
           // Build the list of Source and Sink Channels (should only be 0 or 1 of each)
           val sourceChannel = argsMap.values
             .mapNotNull { it as? GrpcMessageSourceChannel<*> }
             .singleOrNull()
-          val sinkChannel =
-            argsMap.values
-              .mapNotNull { it as? GrpcMessageSinkChannel<*> }
-              .singleOrNull()
+          val sinkChannel = argsMap.values
+            .mapNotNull { it as? GrpcMessageSinkChannel<*> }
+            .singleOrNull()
           // Launch a coroutine for each Source and Sink Channels to bridge the data
           sourceChannel?.let { launch { sourceChannel.bridgeFromSource() } }
           sinkChannel?.let { launch { sinkChannel.bridgeToSink() } }
@@ -63,8 +67,6 @@ internal fun WebAction.asChain(
             sinkChannel?.close()
           }
         }
-      } else {
-        function.callBy(argsMap)
       } ?: throw IllegalStateException("Null return from WebAction")
     }
   }

misk/src/test/kotlin/misk/grpc/GrpcMessageSinkChannelTest.kt

@@ -7,19 +7,15 @@ import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.runTest
 import okio.Buffer
 import okio.ByteString.Companion.decodeHex
-import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.AfterEach
-import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
+import kotlin.test.assertEquals
 import kotlin.time.Duration.Companion.milliseconds
-import kotlin.time.Duration.Companion.seconds
 
 class GrpcMessageSinkChannelTest {
-
   private val buffer = Buffer()
   private val writer = GrpcMessageSink(buffer, HelloRequest.ADAPTER, "identity")
 
-
   @AfterEach
   fun tearDown() {
     writer.close()
@@ -28,28 +24,18 @@ class GrpcMessageSinkChannelTest {
 
   @Test
   fun `test bridge from Channel to GrpcMessageSink HelloRequest`() = runTest {
-
     val channel = Channel<HelloRequest>()
     launch { GrpcMessageSinkChannel(channel, writer).bridgeToSink() }
 
     channel.send(HelloRequest("localhost"))
     eventually(100.milliseconds) {
-      buffer.readByteString().also { byteString ->
-        assertThat(byteString)
-          .isEqualTo("000000000b0a096c6f63616c686f7374".decodeHex())
-      }
+      assertEquals("000000000b0a096c6f63616c686f7374".decodeHex(), buffer.readByteString())
     }
 
     channel.send(HelloRequest("proxy"))
     eventually(100.milliseconds) {
-      buffer.readByteString().also { byteString ->
-        assertThat(byteString)
-          .isEqualTo("00000000070a0570726f7879".decodeHex())
-      }
+      assertEquals("00000000070a0570726f7879".decodeHex(), buffer.readByteString())
     }
     channel.close()
-
   }
-
 }
-

misk/src/test/kotlin/misk/grpc/GrpcMessageSourceChannelTest.kt

@@ -7,13 +7,12 @@ import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.runTest
 import okio.Buffer
 import okio.ByteString.Companion.decodeHex
-import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.Test
+import kotlin.test.assertEquals
 import kotlin.time.Duration.Companion.milliseconds
 
 class GrpcMessageSourceChannelTest {
-
   private val buffer = Buffer()
   private val reader = GrpcMessageSource(buffer, HelloRequest.ADAPTER)
 
@@ -35,17 +34,11 @@ class GrpcMessageSourceChannelTest {
     launch { GrpcMessageSourceChannel(channel, reader, coroutineContext).bridgeFromSource() }
 
     eventually(100.milliseconds) {
-      channel.receive().also { request ->
-        assertThat(request).isEqualTo(HelloRequest("localhost"))
-      }
+      assertEquals(HelloRequest("localhost"), channel.receive())
     }
 
     eventually(100.milliseconds) {
-      channel.receive().also { request ->
-        assertThat(request).isEqualTo(HelloRequest("proxy"))
-      }
+      assertEquals(HelloRequest("proxy"), channel.receive())
     }
   }
-
 }
-

misk/src/test/kotlin/misk/grpc/GrpcSourceSinkTest.kt

@@ -4,24 +4,26 @@ import com.squareup.protos.test.grpc.HelloReply
 import com.squareup.protos.test.grpc.HelloRequest
 import okio.Buffer
 import okio.ByteString.Companion.decodeHex
-import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Test
+import kotlin.test.assertEquals
 
 class GrpcSourceSinkTest {
   @Test
   fun grpcMessageSourceHelloRequest() {
     val buffer = Buffer()
     buffer.write("000000000b0a096c6f63616c686f7374".decodeHex())
     val reader = GrpcMessageSource(buffer, HelloRequest.ADAPTER)
-    assertThat(reader.read()).isEqualTo(HelloRequest("localhost"))
+
+    assertEquals(HelloRequest("localhost"), reader.read())
   }
 
   @Test
   fun grpcMessageSourceHelloReply() {
     val buffer = Buffer()
     buffer.write("00000000110a0f48656c6c6f206c6f63616c686f7374".decodeHex())
     val reader = GrpcMessageSource(buffer, HelloReply.ADAPTER)
-    assertThat(reader.read()).isEqualTo(HelloReply("Hello localhost"))
+
+    assertEquals(HelloReply("Hello localhost"), reader.read())
   }
 
   @Test
@@ -30,8 +32,8 @@ class GrpcSourceSinkTest {
     val writer = GrpcMessageSink(buffer, HelloRequest.ADAPTER, "identity")
     writer.write(HelloRequest("localhost"))
     writer.close()
-    assertThat(buffer.readByteString())
-      .isEqualTo("000000000b0a096c6f63616c686f7374".decodeHex())
+
+    assertEquals("000000000b0a096c6f63616c686f7374".decodeHex(), buffer.readByteString())
   }
 
   @Test
@@ -40,8 +42,7 @@ class GrpcSourceSinkTest {
     val writer = GrpcMessageSink(buffer, HelloReply.ADAPTER, "identity")
     writer.write(HelloReply("Hello localhost"))
     writer.close()
-    assertThat(buffer.readByteString()).isEqualTo(
-      "00000000110a0f48656c6c6f206c6f63616c686f7374".decodeHex()
-    )
+
+    assertEquals("00000000110a0f48656c6c6f206c6f63616c686f7374".decodeHex(), buffer.readByteString())
   }
 }

misk/src/test/kotlin/misk/logging/DynamicMdcContextTest.kt

@@ -3,11 +3,11 @@ package misk.logging
 import io.kotest.assertions.assertSoftly
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
-import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import org.slf4j.MDC
 import wisp.logging.getLogger
+import kotlin.test.assertEquals
 
 class DynamicMdcContextTest {
   @BeforeEach
@@ -33,9 +33,9 @@ class DynamicMdcContextTest {
             logger.info { "level2 added" }
           }
           assertSoftly {
-            assertThat(MDC.get("level0")).isEqualTo("value0")
-            assertThat(MDC.get("level1")).isEqualTo("value1")
-            assertThat(MDC.get("level2")).isEqualTo("value2")
+            assertEquals("value0", MDC.get("level0"))
+            assertEquals("value1", MDC.get("level1"))
+            assertEquals("value2", MDC.get("level2"))
           }
         }
       }