From b7ac23cb54a661c95365ba2ffd26b6312f7bd8d6 Mon Sep 17 00:00:00 2001
From: Oliver Foggin <oliverfoggin@me.com>
Date: Thu, 14 Dec 2023 18:56:45 +0000
Subject: [PATCH] Add transform function

---
 .../SubscriberReducer.swift                   | 43 ++++++++++++++-----
 1 file changed, 32 insertions(+), 11 deletions(-)

diff --git a/Sources/ComposableSubscriber/SubscriberReducer.swift b/Sources/ComposableSubscriber/SubscriberReducer.swift
index e9eda5f..01bc45f 100644
--- a/Sources/ComposableSubscriber/SubscriberReducer.swift
+++ b/Sources/ComposableSubscriber/SubscriberReducer.swift
@@ -1,21 +1,37 @@
 import ComposableArchitecture
 
 extension Reducer {
-	public func subscribe<TriggerAction, T>(
-		to stream: @escaping () async -> AsyncStream<T>,
+	public func subscribe<TriggerAction, StreamElement>(
+		to stream: @escaping () async -> AsyncStream<StreamElement>,
 		on triggerAction: CaseKeyPath<Action, TriggerAction>,
-		with responseAction: CaseKeyPath<Action, T>
-	) -> _SubscribeReducer<Self, TriggerAction, T> {
+		with responseAction: CaseKeyPath<Action, StreamElement>
+	) -> _SubscribeReducer<Self, TriggerAction, StreamElement, StreamElement> {
 		.init(
 			parent: self,
 			on: triggerAction,
 			to: stream,
-			with: responseAction
+			with: responseAction,
+			transform: { $0 }
+		)
+	}
+
+	public func subscribe<TriggerAction, StreamElement, Value>(
+		to stream: @escaping () async -> AsyncStream<StreamElement>,
+		on triggerAction: CaseKeyPath<Action, TriggerAction>,
+		with responseAction: CaseKeyPath<Action, Value>,
+		transform: @escaping (StreamElement) -> Value
+	) -> _SubscribeReducer<Self, TriggerAction, StreamElement, Value> {
+		.init(
+			parent: self,
+			on: triggerAction,
+			to: stream,
+			with: responseAction,
+			transform: transform
 		)
 	}
 }
 
-public struct _SubscribeReducer<Parent: Reducer, TriggerAction, T>: Reducer {
+public struct _SubscribeReducer<Parent: Reducer, TriggerAction, StreamElement, Value>: Reducer {
 	@usableFromInline
 	let parent: Parent
 
@@ -23,21 +39,26 @@ public struct _SubscribeReducer<Parent: Reducer, TriggerAction, T>: Reducer {
 	let triggerAction: AnyCasePath<Parent.Action, TriggerAction>
 
 	@usableFromInline
-	let stream: () async -> AsyncStream<T>
+	let stream: () async -> AsyncStream<StreamElement>
 
 	@usableFromInline
-	let responseAction: AnyCasePath<Parent.Action, T>
+	let responseAction: AnyCasePath<Parent.Action, Value>
+
+	@usableFromInline
+	let transform: (StreamElement) -> Value
 
 	init(
 		parent: Parent,
 		on triggerAction: CaseKeyPath<Parent.Action, TriggerAction>,
-		to stream: @escaping () async -> AsyncStream<T>,
-		with responseAction: CaseKeyPath<Parent.Action, T>
+		to stream: @escaping () async -> AsyncStream<StreamElement>,
+		with responseAction: CaseKeyPath<Parent.Action, Value>,
+		transform: @escaping (StreamElement) -> Value
 	) {
 		self.parent = parent
 		self.triggerAction = AnyCasePath(triggerAction)
 		self.stream = stream
 		self.responseAction = AnyCasePath(responseAction)
+		self.transform = transform
 	}
 
 	public func reduce(into state: inout Parent.State, action: Parent.Action) -> Effect<Parent.Action> {
@@ -51,7 +72,7 @@ public struct _SubscribeReducer<Parent: Reducer, TriggerAction, T>: Reducer {
 			effects,
 			.run { send in
 				for await value in await stream() {
-					await send(responseAction.embed(value))
+					await send(responseAction.embed(transform(value)))
 				}
 			}
 		)