Merge pull request #138 from surfstudio/SPT-1998-hide-async-stream

SPT-1998 Скрытие AsyncStream и исправление опечаток (6)

Example/Example.xcodeproj/project.pbxproj

@@ -3,7 +3,7 @@
 	archiveVersion = 1;
 	classes = {
 	};
-	objectVersion = 60;
+	objectVersion = 54;
 	objects = {
 
 /* Begin PBXBuildFile section */
@@ -350,7 +350,6 @@
 			packageReferences = (
 				50C445A02BC81FB300C515E6 /* XCRemoteSwiftPackageReference "SwiftMessages" */,
 				50C445A62BC8202600C515E6 /* XCRemoteSwiftPackageReference "ReactiveDataDisplayManager" */,
-				5097DF472BCD62AA00D422EE /* XCLocalSwiftPackageReference "../NodeKit" */,
 				50B5697E2BD183F00054DC09 /* XCRemoteSwiftPackageReference "iOS-Utils" */,
 				50B569812BD184150054DC09 /* XCRemoteSwiftPackageReference "Nuke" */,
 				50B5698A2BD18F040054DC09 /* XCRemoteSwiftPackageReference "SnapKit" */,
@@ -603,13 +602,6 @@
 		};
 /* End XCConfigurationList section */
 
-/* Begin XCLocalSwiftPackageReference section */
-		5097DF472BCD62AA00D422EE /* XCLocalSwiftPackageReference "../NodeKit" */ = {
-			isa = XCLocalSwiftPackageReference;
-			relativePath = ../NodeKit;
-		};
-/* End XCLocalSwiftPackageReference section */
-
 /* Begin XCRemoteSwiftPackageReference section */
 		50B5697E2BD183F00054DC09 /* XCRemoteSwiftPackageReference "iOS-Utils" */ = {
 			isa = XCRemoteSwiftPackageReference;

Example/Modules/MockServer/Package.swift

@@ -14,7 +14,7 @@ let package = Package(
         )
     ],
     dependencies: [
-        .package(path: "../../../NodeKit"),
+        .package(path: "../../.."),
         .package(path: "../Models")
     ],
     targets: [

Example/Modules/Models/Package.swift

@@ -14,7 +14,7 @@ let package = Package(
         )
     ],
     dependencies: [
-        .package(path: "../../../NodeKit")
+        .package(path: "../../..")
     ],
     targets: [
         .target(

Example/Modules/Services/Package.swift

@@ -15,14 +15,16 @@ let package = Package(
     ],
     dependencies: [
         .package(path: "../Models"),
-        .package(path: "../../../NodeKit")
+        .package(path: "../MockServer"),
+        .package(path: "../../..")
     ],
     targets: [
         .target(
             name: "Services",
             dependencies: [
                 "Models",
-                "NodeKit"
+                "NodeKit",
+                "MockServer"
             ],
             path: "Services"
         )

NodeKit/NodeKit.xcodeproj/project.pbxproj

@@ -133,7 +133,6 @@
 		50B569782BCFEF610054DC09 /* AsyncStreamNodeResultPublisher.swift in Sources */ = {isa = PBXBuildFile; fileRef = 50B569772BCFEF610054DC09 /* AsyncStreamNodeResultPublisher.swift */; };
 		50B5697A2BCFF1A10054DC09 /* CombineCompatibleNode.swift in Sources */ = {isa = PBXBuildFile; fileRef = 50B569792BCFF1A10054DC09 /* CombineCompatibleNode.swift */; };
 		50B6838F2BBF3615001F7EA3 /* AccessSafeNodeTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 50B6838E2BBF3615001F7EA3 /* AccessSafeNodeTests.swift */; };
-		50BEF7BE2BE8A713003DCB04 /* MergedAsyncStreamNode.swift in Sources */ = {isa = PBXBuildFile; fileRef = 50BEF7BD2BE8A713003DCB04 /* MergedAsyncStreamNode.swift */; };
 		50C57AB72BE3871D004C344E /* ServiceChainProvider.swift in Sources */ = {isa = PBXBuildFile; fileRef = 50C57AB62BE3871D004C344E /* ServiceChainProvider.swift */; };
 		50C57AB92BE388C6004C344E /* ChainBuilder.swift in Sources */ = {isa = PBXBuildFile; fileRef = 50C57AB82BE388C6004C344E /* ChainBuilder.swift */; };
 		50C8EB302BBD90DE00C5CB93 /* CombineNodeTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 50C8EB2F2BBD90DE00C5CB93 /* CombineNodeTests.swift */; };
@@ -365,7 +364,6 @@
 		50B569772BCFEF610054DC09 /* AsyncStreamNodeResultPublisher.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = AsyncStreamNodeResultPublisher.swift; sourceTree = "<group>"; };
 		50B569792BCFF1A10054DC09 /* CombineCompatibleNode.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CombineCompatibleNode.swift; sourceTree = "<group>"; };
 		50B6838E2BBF3615001F7EA3 /* AccessSafeNodeTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = AccessSafeNodeTests.swift; sourceTree = "<group>"; };
-		50BEF7BD2BE8A713003DCB04 /* MergedAsyncStreamNode.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = MergedAsyncStreamNode.swift; sourceTree = "<group>"; };
 		50C57AB62BE3871D004C344E /* ServiceChainProvider.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = ServiceChainProvider.swift; sourceTree = "<group>"; };
 		50C57AB82BE388C6004C344E /* ChainBuilder.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = ChainBuilder.swift; sourceTree = "<group>"; };
 		50C8EB272BBD7A2200C5CB93 /* AsyncStreamCombineNode.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = AsyncStreamCombineNode.swift; sourceTree = "<group>"; };
@@ -517,7 +515,6 @@
 				501349142BE3E6D2002CC3DA /* AnyAsyncNode.swift */,
 				5005EB222BB8A6D900B670CD /* AsyncStreamNode.swift */,
 				501349162BE3E770002CC3DA /* AnyAsyncStreamNode.swift */,
-				50BEF7BD2BE8A713003DCB04 /* MergedAsyncStreamNode.swift */,
 			);
 			path = Async;
 			sourceTree = "<group>";
@@ -1521,7 +1518,6 @@
 				5005EB212BB8A6D000B670CD /* AsyncNode.swift in Sources */,
 				90B609A5283E1287006F4309 /* ResponseProcessorNode.swift in Sources */,
 				90B60995283E1287006F4309 /* DTOEncoderNode.swift in Sources */,
-				50BEF7BE2BE8A713003DCB04 /* MergedAsyncStreamNode.swift in Sources */,
 				90B60914283E1268006F4309 /* ETagConstants.swift in Sources */,
 				50528E292BAE162600E86CB6 /* LoggerStreamNode.swift in Sources */,
 				90B609A2283E1287006F4309 /* ResponseDataParserNode.swift in Sources */,

NodeKit/NodeKit/CacheNode/FirstCachePolicyNode.swift

@@ -13,29 +13,29 @@ import Foundation
 /// - SeeAlso: `FirstCachePolicyNode`
 ///
 /// - cantGetURLRequest: Возникает в случае, если запрос отправленный в сеть не содержит `URLRequest`
-public enum BaseFirstCachePolicyNodeError: Error {
+enum BaseFirstCachePolicyNodeError: Error {
     case cantGetURLRequest
 }
 
 /// Этот узел реализует политику кэширования
 /// "Сначала читаем из кэша, а затем запрашиваем у сервера"
 /// - Important: В общем случае слушатель может быть оповещен дважды. Первый раз, когда ответ прочитан из кэша, а второй раз, когда он был получен с сервера.
-open class FirstCachePolicyNode: AsyncStreamNode {
+class FirstCachePolicyNode: AsyncStreamNode {
     // MARK: - Nested
 
     /// Тип для читающего из URL кэша узла
-    public typealias CacheReaderNode = AsyncNode<URLNetworkRequest, Json>
+    typealias CacheReaderNode = AsyncNode<URLNetworkRequest, Json>
 
     /// Тип для следующего узла
-    public typealias NextProcessorNode = AsyncNode<RawURLRequest, Json>
+    typealias NextProcessorNode = AsyncNode<RawURLRequest, Json>
 
     // MARK: - Properties
 
     /// Следующий узел для обработки.
-    public var next: any NextProcessorNode
+    var next: any NextProcessorNode
 
     /// Узел для чтения из кэша.
-    public var cacheReaderNode: any CacheReaderNode
+    var cacheReaderNode: any CacheReaderNode
 
     // MARK: - Init and Deinit
 
@@ -44,7 +44,7 @@ open class FirstCachePolicyNode: AsyncStreamNode {
     /// - Parameters:
     ///   - cacheReaderNode: Узел для чтения из кэша.
     ///   - next: Следующий узел для обработки.
-    public init(cacheReaderNode: any CacheReaderNode, next: any NextProcessorNode) {
+    init(cacheReaderNode: any CacheReaderNode, next: any NextProcessorNode) {
         self.cacheReaderNode = cacheReaderNode
         self.next = next
     }
@@ -55,12 +55,12 @@ open class FirstCachePolicyNode: AsyncStreamNode {
     /// а затем, передает управление следующему узлу.
     /// В случае, если получить `URLRequest` не удалось,
     /// то управление просто передается следующему узлу
-    public func process(
+    func process(
         _ data: RawURLRequest,
         logContext: LoggingContextProtocol
     ) -> AsyncStream<NodeResult<Json>> {
         return AsyncStream { continuation in
-            Task {
+            let task = Task {
                 if let request = data.toURLRequest() {
                     let cacheResult = await cacheReaderNode.process(request, logContext: logContext)
                     continuation.yield(cacheResult)
@@ -70,6 +70,9 @@ open class FirstCachePolicyNode: AsyncStreamNode {
                 continuation.yield(nextResult)
                 continuation.finish()
             }
+            continuation.onTermination = { _ in
+                task.cancel()
+            }
         }
     }
 }

NodeKit/NodeKit/Core/Node/Async/AnyAsyncStreamNode.swift

@@ -8,7 +8,7 @@
 
 import Foundation
 
-public struct AnyAsyncStreamNode<Input, Output>: AsyncStreamNode {
+struct AnyAsyncStreamNode<Input, Output>: AsyncStreamNode {
 
     // MARK: - Private Properties
 
@@ -22,7 +22,7 @@ public struct AnyAsyncStreamNode<Input, Output>: AsyncStreamNode {
 
     // MARK: - AsyncNode
 
-    public func process(_ data: Input, logContext: LoggingContextProtocol) -> AsyncStream<NodeResult<Output>> {
+    func process(_ data: Input, logContext: LoggingContextProtocol) -> AsyncStream<NodeResult<Output>> {
         return node.process(data, logContext: logContext)
     }
 }

NodeKit/NodeKit/Core/Node/Async/AsyncNode.swift

@@ -11,9 +11,9 @@ import Foundation
 
 /// Протокол ноды, описывающий подход преобразования входных данных в результат с помощью SwiftConcurrency.
 /// Поддерживает обработку результатов с помощью Combine, наследуя протокол ``CombineCompatibleNode``.
-/// Содержит параметры для логов, наследуя протокол ``LoggableNode``.
+/// Содержит параметры для логов, наследуя протокол ``Node``.
 /// Применим для узлов, которые возвращают один результат.
-public protocol AsyncNode<Input, Output>: LoggableNode, CombineCompatibleNode<Self.Input, Self.Output> {
+public protocol AsyncNode<Input, Output>: Node, CombineCompatibleNode<Self.Input, Self.Output> {
     associatedtype Input
     associatedtype Output
 
@@ -29,12 +29,6 @@ public protocol AsyncNode<Input, Output>: LoggableNode, CombineCompatibleNode<Se
     ///
     /// - Returns: Cтруктура-обертку текущей ноды ``AnyAsyncNode``.
     func eraseToAnyNode() -> AnyAsyncNode<Input, Output>
-
-    /// Метод, позволяющий объединить две ноды с одинаковыми Input и Output в AsyncStreamNode.
-    ///
-    /// - Parameter node: Нода, необходимая для объединения.
-    /// - Returns: Нода AsyncStreamNode, включающая текущую и переданную ноду.
-    func merged(with node: any AsyncNode<Input, Output>) -> any AsyncStreamNode<Input, Output>
 }
 
 public extension AsyncNode {
@@ -74,13 +68,7 @@ public extension AsyncNode {
         return AnyAsyncNode(node: self)
     }
 
-    /// Стандартная реализация объединения двух узлов в AsyncStreamNode.
-    ///
-    /// - Parameter node: Нода, необходимая для объединения.
-    /// - Returns: Нода AsyncStreamNode, включающая текущую и переданную ноду.
-    func merged(with node: any AsyncNode<Input, Output>) -> any AsyncStreamNode<Input, Output> {
-        return MergedAsyncStreamNode(firstNode: self, secondNode: node)
-    }
+
 }
 
 /// Содержит синтаксический сахар для работы с узлами, у которых входящий тип = `Void`

NodeKit/NodeKit/Core/Node/Async/AsyncStreamNode.swift

@@ -11,9 +11,9 @@ import Foundation
 
 /// Протокол ноды, описывающий подход преобразования входных данных в результат с помощью SwiftConcurrency.
 /// Поддерживает обработку результатов с помощью Combine, наследуя протокол ``CombineCompatibleNode``.
-/// Содержит параметры для логов, наследуя протокол ``LoggableNode``.
+/// Содержит параметры для логов, наследуя протокол ``Node``.
 /// Применим для узлов, которые могут вернуть несколько результатов
-public protocol AsyncStreamNode<Input, Output>: LoggableNode, CombineCompatibleNode<Self.Input, Self.Output> {
+protocol AsyncStreamNode<Input, Output>: Node, CombineCompatibleNode<Self.Input, Self.Output> {
     associatedtype Input
     associatedtype Output
 
@@ -31,7 +31,7 @@ public protocol AsyncStreamNode<Input, Output>: LoggableNode, CombineCompatibleN
     func eraseToAnyNode() -> AnyAsyncStreamNode<Input, Output>
 }
 
-public extension AsyncStreamNode {
+extension AsyncStreamNode {
 
     /// Метод process с созданием нового лог контекста.
     @discardableResult
@@ -70,7 +70,7 @@ public extension AsyncStreamNode {
 }
 
 /// Содержит синтаксический сахар для работы с узлами, у которых входящий тип = `Void`
-public extension AsyncStreamNode where Input == Void {
+extension AsyncStreamNode where Input == Void {
 
     /// Вызывает `process(_:)`
     @discardableResult

NodeKit/NodeKit/Core/Node/Node.swift

@@ -1,5 +1,5 @@
 //
-//  LoggableNode.swift
+//  Node.swift
 //  CoreNetKitWithExample
 //
 //  Created by Александр Кравченков on 27/11/2018.
@@ -10,16 +10,16 @@ import Foundation
 
 /// Протокол, описывающий любой узел или цепочку узлов.
 /// Необходим для объединения всех типов узлов и добавления общих методов.
-public protocol LoggableNode { }
+public protocol Node { }
 
 /// Содержит вычисляемые константы
-public extension LoggableNode {
+public extension Node {
     /// Возвращает имя типа строкой
     var objectName: String {
         return "\(type(of: self))"
     }
 
-    /// Имея обхекта в формате:
+    /// Имея объекта в формате:
     /// <<<===\(self.objectName)===>>>" + `String.lineTabDeilimeter`
     var logViewObjectName: String {
         return "<<<===\(self.objectName)===>>>" + .lineTabDeilimeter

NodeKit/NodeKit/Utils/Logging/LoggerStreamNode.swift

@@ -8,29 +8,29 @@
 
 /// Этот узел выполняет выведение лога в консоль.
 /// Сразу же передает управление следующему узлу и подписывается на выполнение операций.
-open class LoggerStreamNode<Input, Output>: AsyncStreamNode {
+class LoggerStreamNode<Input, Output>: AsyncStreamNode {
 
     /// Следующий узел для обработки.
-    open var next: any AsyncStreamNode<Input, Output>
+    var next: any AsyncStreamNode<Input, Output>
     /// Содержит список ключей, по которым будет отфлитрован лог.
-    open var filters: [String]
+    var filters: [String]
 
     /// Инициаллизирует объект.
     ///
     /// - Parameters:
     ///   - next: Следующий узел для обработки.
     ///   - filters: Содержит список ключей, по которым будет отфлитрован лог.
-    public init(next: any AsyncStreamNode<Input, Output>, filters: [String] = []) {
+    init(next: any AsyncStreamNode<Input, Output>, filters: [String] = []) {
         self.next = next
         self.filters = filters
     }
 
     /// Сразу же передает управление следующему узлу и подписывается на выполнение операций.
     ///
     /// - Parameter data: Данные для обработки. Этот узел их не импользует.
-    public func process(_ data: Input, logContext: LoggingContextProtocol) -> AsyncStream<NodeResult<Output>> {
+    func process(_ data: Input, logContext: LoggingContextProtocol) -> AsyncStream<NodeResult<Output>> {
         return AsyncStream { continuation in
-            Task {
+            let task = Task {
                 for await result in next.process(data, logContext: logContext) {
                     continuation.yield(result)
                 }
@@ -40,6 +40,9 @@ open class LoggerStreamNode<Input, Output>: AsyncStreamNode {
                     .forEach { print($0.description) }
                 continuation.finish()
             }
+            continuation.onTermination = { _ in
+                task.cancel()
+            }
         }
     }
 }