remove autobahn docker check

DitchOoM · Jun 11, 2024 · 464e036 · 464e036
1 parent ae95329
commit 464e036
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 4 deletions.
diff --git a/.github/workflows/review.yaml b/.github/workflows/review.yaml
@@ -26,10 +26,10 @@ jobs:
         uses: gradle/actions/setup-gradle@v3
       - name: Tests with Gradle Major
         if:  ${{ contains(github.event.pull_request.labels.*.name, 'major') }}
-        run: ./gradlew startAutobahnDockerContainer ktlintCheck assemble build check allTests publishToMavenLocal -PincrementMajor=true
+        run: ./gradlew ktlintCheck assemble build check allTests publishToMavenLocal -PincrementMajor=true
       - name: Tests with Gradle Minor
         if:  ${{ contains(github.event.pull_request.labels.*.name, 'minor') }}
-        run: ./gradlew startAutobahnDockerContainer ktlintCheck assemble build check allTests publishToMavenLocal -PincrementMinor=true
+        run: ./gradlew ktlintCheck assemble build check allTests publishToMavenLocal -PincrementMinor=true
       - name: Tests with Gradle Patch
         if:  ${{ !contains(github.event.pull_request.labels.*.name, 'major') && !contains(github.event.pull_request.labels.*.name, 'minor') }}
-        run: ./gradlew startAutobahnDockerContainer ktlintCheck assemble build check allTests publishToMavenLocal
+        run: ./gradlew ktlintCheck assemble build check allTests publishToMavenLocal
diff --git a/mqtt-client/src/commonMain/kotlin/com/ditchoom/mqtt/client/ControlPacketOperation.kt b/mqtt-client/src/commonMain/kotlin/com/ditchoom/mqtt/client/ControlPacketOperation.kt
@@ -9,6 +9,8 @@ import com.ditchoom.mqtt.controlpacket.ISubscription
 import com.ditchoom.mqtt.controlpacket.IUnsubscribeAcknowledgment
 import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.FlowCollector
+import kotlinx.coroutines.flow.combine
 
 sealed interface PublishOperation {
     object QoSAtMostOnceComplete : PublishOperation
@@ -38,6 +40,13 @@ data class SubscribeOperation(
     val packetId: Int,
     val subscriptions: Map<ISubscription, Flow<IPublishMessage>>,
     val subAck: Deferred<ISubscribeAcknowledgement>,
-)
+): Flow<IPublishMessage> {
+    override suspend fun collect(collector: FlowCollector<IPublishMessage>) {
+        combine(subscriptions.values.asIterable()) { array ->
+            array.forEach { collector.emit(it) }
+        }
+    }
+
+}
 
 data class UnsubscribeOperation(val packetId: Int, val unsubAck: Deferred<IUnsubscribeAcknowledgment>)