Subscribe & Presence Event Engine

Examples/Sources/MasterDetailTableViewController.swift

@@ -260,6 +260,8 @@ class MasterDetailTableViewController: UITableViewController {
             print("Status disconnected")
           case .disconnectedUnexpectedly:
             print("Status disconnected unexpectedly!")
+          case .connectionError:
+            print("Cannot establish initial conection to the remote system")
           }
         case let .subscriptionChanged(subscribeChange):
           switch subscribeChange {

Podfile.lock

@@ -19,4 +19,4 @@ SPEC CHECKSUMS:
 
 PODFILE CHECKSUM: 61a40240486621bb01f596fdd5bc632504940fab
 
-COCOAPODS: 1.12.1
+COCOAPODS: 1.14.3

PubNubMembership/Sources/Membership+PubNub.swift

@@ -18,7 +18,6 @@ import PubNubUser
 public protocol PubNubMembershipInterface {
   /// A copy of the configuration object used for this session
   var configuration: PubNubConfiguration { get }
-
   /// Session used for performing request/response REST calls
   var networkSession: SessionReplaceable { get }
 
@@ -268,6 +267,7 @@ public extension PubNubMembershipInterface {
     (requestConfig.customSession ?? networkSession)
       .route(
         router,
+        requestOperator: configuration.automaticRetry?[.appContext],
         responseDecoder: FetchMultipleValueResponseDecoder<PubNubMembership.PartialSpace>(),
         responseQueue: requestConfig.responseQueue
       ) { result in
@@ -320,6 +320,7 @@ public extension PubNubMembershipInterface {
     (requestConfig.customSession ?? networkSession)
       .route(
         router,
+        requestOperator: configuration.automaticRetry?[.appContext],
         responseDecoder: FetchMultipleValueResponseDecoder<PubNubMembership.PartialUser>(),
         responseQueue: requestConfig.responseQueue
       ) { result in
@@ -365,6 +366,7 @@ public extension PubNubMembershipInterface {
     (requestConfig.customSession ?? networkSession)
       .route(
         router,
+        requestOperator: configuration.automaticRetry?[.appContext],
         responseDecoder: FetchStatusResponseDecoder(),
         responseQueue: requestConfig.responseQueue
       ) { result in
@@ -401,6 +403,7 @@ public extension PubNubMembershipInterface {
     (requestConfig.customSession ?? networkSession)
       .route(
         router,
+        requestOperator: configuration.automaticRetry?[.appContext],
         responseDecoder: FetchStatusResponseDecoder(),
         responseQueue: requestConfig.responseQueue
       ) { result in
@@ -463,6 +466,7 @@ public extension PubNubMembershipInterface {
     (requestConfig.customSession ?? networkSession)
       .route(
         router,
+        requestOperator: configuration.automaticRetry?[.appContext],
         responseDecoder: FetchStatusResponseDecoder(),
         responseQueue: requestConfig.responseQueue
       ) { result in
@@ -499,6 +503,7 @@ public extension PubNubMembershipInterface {
     (requestConfig.customSession ?? networkSession)
       .route(
         router,
+        requestOperator: configuration.automaticRetry?[.appContext],
         responseDecoder: FetchStatusResponseDecoder(),
         responseQueue: requestConfig.responseQueue
       ) { result in

PubNubSpace/Sources/Space+PubNub.swift

@@ -16,7 +16,6 @@ import PubNub
 public protocol PubNubSpaceInterface {
   /// A copy of the configuration object used for this session
   var configuration: PubNubConfiguration { get }
-
   /// Session used for performing request/response REST calls
   var networkSession: SessionReplaceable { get }
 
@@ -213,6 +212,7 @@ public extension PubNubSpaceInterface {
     (requestConfig.customSession ?? networkSession)
       .route(
         router,
+        requestOperator: configuration.automaticRetry?[.appContext],
         responseDecoder: FetchMultipleValueResponseDecoder<PubNubSpace>(),
         responseQueue: requestConfig.responseQueue
       ) { result in
@@ -237,6 +237,7 @@ public extension PubNubSpaceInterface {
     (requestConfig.customSession ?? networkSession)
       .route(
         router,
+        requestOperator: configuration.automaticRetry?[.appContext],
         responseDecoder: FetchSingleValueResponseDecoder<PubNubSpace>(),
         responseQueue: requestConfig.responseQueue
       ) { result in
@@ -273,6 +274,7 @@ public extension PubNubSpaceInterface {
     (requestConfig.customSession ?? networkSession)
       .route(
         router,
+        requestOperator: configuration.automaticRetry?[.appContext],
         responseDecoder: FetchSingleValueResponseDecoder<PubNubSpace>(),
         responseQueue: requestConfig.responseQueue
       ) { result in
@@ -317,6 +319,7 @@ public extension PubNubSpaceInterface {
     (requestConfig.customSession ?? networkSession)
       .route(
         router,
+        requestOperator: configuration.automaticRetry?[.appContext],
         responseDecoder: FetchStatusResponseDecoder(),
         responseQueue: requestConfig.responseQueue
       ) { result in

PubNubUser/Sources/User+PubNub.swift

@@ -9,14 +9,12 @@
 //
 
 import Foundation
-
 import PubNub
 
 /// Protocol interface to manage `PubNubUser` entities using closures
 public protocol PubNubUserInterface {
   /// A copy of the configuration object used for this session
   var configuration: PubNubConfiguration { get }
-
   /// Session used for performing request/response REST calls
   var networkSession: SessionReplaceable { get }
 
@@ -221,6 +219,7 @@ public extension PubNubUserInterface {
     (requestConfig.customSession ?? networkSession)?
       .route(
         router,
+        requestOperator: configuration.automaticRetry?[.appContext],
         responseDecoder: FetchMultipleValueResponseDecoder<PubNubUser>(),
         responseQueue: requestConfig.responseQueue
       ) { result in
@@ -248,6 +247,7 @@ public extension PubNubUserInterface {
     (requestConfig.customSession ?? networkSession)
       .route(
         router,
+        requestOperator: configuration.automaticRetry?[.appContext],
         responseDecoder: FetchSingleValueResponseDecoder<PubNubUser>(),
         responseQueue: requestConfig.responseQueue
       ) {
@@ -288,6 +288,7 @@ public extension PubNubUserInterface {
     (requestConfig.customSession ?? networkSession)
       .route(
         router,
+        requestOperator: configuration.automaticRetry?[.appContext],
         responseDecoder: FetchSingleValueResponseDecoder<PubNubUser>(),
         responseQueue: requestConfig.responseQueue
       ) { result in
@@ -336,6 +337,7 @@ public extension PubNubUserInterface {
     (requestConfig.customSession ?? networkSession)
       .route(
         router,
+        requestOperator: configuration.automaticRetry?[.appContext],
         responseDecoder: FetchStatusResponseDecoder(),
         responseQueue: requestConfig.responseQueue
       ) { result in

Sources/PubNub/APIs/File+PubNub.swift

@@ -29,6 +29,7 @@ public extension PubNub {
   ) {
     route(
       FileManagementRouter(.list(channel: channel, limit: limit, next: next), configuration: configuration),
+      requestOperator: configuration.automaticRetry?[.files],
       responseDecoder: FileListResponseDecoder(),
       custom: requestConfig
     ) { result in
@@ -60,6 +61,7 @@ public extension PubNub {
   ) {
     route(
       FileManagementRouter(.delete(channel: channel, fileId: fileId, filename: filename), configuration: configuration),
+      requestOperator: configuration.automaticRetry?[.files],
       responseDecoder: FileGeneralSuccessResponseDecoder(),
       custom: requestConfig
     ) { result in
@@ -137,6 +139,7 @@ public extension PubNub {
         .generateURL(channel: channel, body: .init(name: remoteFilename)),
         configuration: configuration
       ),
+      requestOperator: configuration.automaticRetry?[.files],
       responseDecoder: FileGenerateResponseDecoder(),
       custom: requestConfig
     ) { [configuration] result in
@@ -225,9 +228,12 @@ public extension PubNub {
       configuration: configuration
     )
 
-    route(router,
-          responseDecoder: PublishResponseDecoder(),
-          custom: request.customRequestConfig) { result in
+    route(
+      router,
+      requestOperator: configuration.automaticRetry?[.files],
+      responseDecoder: PublishResponseDecoder(),
+      custom: request.customRequestConfig
+    ) { result in
       completion?(result.map { $0.payload.timetoken })
     }
   }

Sources/PubNub/EventEngine/Core/Dispatcher.swift

@@ -0,0 +1,113 @@
+//
+//  Dispatcher.swift
+//
+//  Copyright (c) PubNub Inc.
+//  All rights reserved.
+//
+//  This source code is licensed under the license found in the
+//  LICENSE file in the root directory of this source tree.
+//
+
+import Foundation
+
+// MARK: - DispatcherListener
+
+struct DispatcherListener<Event> {
+  let onAnyInvocationCompleted: (([Event]) -> Void)
+}
+
+// MARK: - Dispatcher
+
+protocol Dispatcher<Invocation, Event, Dependencies> {
+  associatedtype Invocation: AnyEffectInvocation
+  associatedtype Event
+  associatedtype Dependencies
+
+  func dispatch(
+    invocations: [EffectInvocation<Invocation>],
+    with dependencies: EventEngineDependencies<Dependencies>,
+    notify listener: DispatcherListener<Event>
+  )
+}
+
+// MARK: - EffectDispatcher
+
+class EffectDispatcher<Invocation: AnyEffectInvocation, Event, Dependencies>: Dispatcher {
+  private let factory: any EffectHandlerFactory<Invocation, Event, Dependencies>
+  private let effectsCache = EffectsCache<Event>()
+
+  init(factory: some EffectHandlerFactory<Invocation, Event, Dependencies>) {
+    self.factory = factory
+  }
+
+  func hasPendingInvocation(_ invocation: Invocation) -> Bool {
+    effectsCache.hasPendingEffect(with: invocation.id)
+  }
+
+  func dispatch(
+    invocations: [EffectInvocation<Invocation>],
+    with dependencies: EventEngineDependencies<Dependencies>,
+    notify listener: DispatcherListener<Event>
+  ) {
+    invocations.forEach {
+      switch $0 {
+      case .managed(let invocation):
+        executeEffect(
+          effect: factory.effect(for: invocation, with: dependencies),
+          storageId: invocation.id,
+          notify: listener
+        )
+      case .regular(let invocation):
+        executeEffect(
+          effect: factory.effect(for: invocation, with: dependencies),
+          storageId: UUID().uuidString,
+          notify: listener
+        )
+      case .cancel(let cancelInvocation):
+        effectsCache.getEffect(with: cancelInvocation.id)?.cancelTask()
+        effectsCache.removeEffect(id: cancelInvocation.id)
+      }
+    }
+  }
+
+  private func executeEffect(
+    effect: some EffectHandler<Event>,
+    storageId id: String,
+    notify listener: DispatcherListener<Event>
+  ) {
+    effectsCache.put(effect: effect, with: id)
+    effect.performTask { [weak effectsCache] results in
+      effectsCache?.removeEffect(id: id)
+      listener.onAnyInvocationCompleted(results)
+    }
+  }
+}
+
+// MARK: - EffectsCache
+
+fileprivate class EffectsCache<Event> {
+  private var managedEffects: Atomic<[String: EffectWrapper<Event>]> = Atomic([:])
+
+  func hasPendingEffect(with id: String) -> Bool {
+    managedEffects.lockedRead { $0[id] } != nil
+  }
+
+  func put(effect: some EffectHandler<Event>, with id: String) {
+    managedEffects.lockedWrite { $0[id] = EffectWrapper<Event>(id: id, effect: effect) }
+  }
+
+  func getEffect(with id: String) -> (any EffectHandler<Event>)? {
+    managedEffects.lockedRead() { $0[id] }?.effect
+  }
+
+  func removeEffect(id: String) {
+    managedEffects.lockedWrite { $0[id] = nil }
+  }
+}
+
+// MARK: - EffectWrapper
+
+fileprivate struct EffectWrapper<Action> {
+  let id: String
+  let effect: any EffectHandler<Action>
+}

Sources/PubNub/EventEngine/Core/EffectHandler.swift

@@ -0,0 +1,76 @@
+//
+//  EffectHandler.swift
+//
+//  Copyright (c) PubNub Inc.
+//  All rights reserved.
+//
+//  This source code is licensed under the license found in the
+//  LICENSE file in the root directory of this source tree.
+//
+import Foundation
+
+// MARK: - EffectHandlerFactory
+
+protocol EffectHandlerFactory<Invocation, Event, Dependencies> {
+  associatedtype Invocation
+  associatedtype Event
+  associatedtype Dependencies
+
+  func effect(
+    for invocation: Invocation,
+    with dependencies: EventEngineDependencies<Dependencies>
+  ) -> any EffectHandler<Event>
+}
+
+// MARK: - EffectHandler
+
+protocol EffectHandler<Event> {
+  associatedtype Event
+
+  func performTask(completionBlock: @escaping ([Event]) -> Void)
+  func cancelTask()
+}
+
+extension EffectHandler {
+  func cancelTask() {}
+}
+
+// MARK: - Delayed Effect Handler
+
+protocol DelayedEffectHandler: AnyObject, EffectHandler {
+  var workItem: DispatchWorkItem? { get set }
+
+  func delayInterval() -> TimeInterval?
+  func onEmptyInterval(notify completionBlock: @escaping ([Event]) -> Void)
+  func onDelayExpired(notify completionBlock: @escaping ([Event]) -> Void)
+}
+
+// MARK: - TimerEffect
+
+class TimerEffect: EffectHandler {
+  private let interval: TimeInterval
+  private var workItem: DispatchWorkItem?
+
+  init?(interval: TimeInterval?) {
+    if let interval = interval {
+      self.interval = interval
+    } else {
+      return nil
+    }
+  }
+
+  func performTask(completionBlock: @escaping ([Void]) -> Void) {
+    let workItem = DispatchWorkItem() {
+      completionBlock([])
+    }
+    DispatchQueue.global(qos: .default).asyncAfter(
+      deadline: .now() + interval,
+      execute: workItem
+    )
+    self.workItem = workItem
+  }
+
+  func cancelTask() {
+    workItem?.cancel()
+  }
+}