handle close and clear

Author: arnauddorgans

Sources/OpenSeaSwift/Internal/Phoenix/PhoenixInternalEvent.swift

@@ -5,14 +5,14 @@ import Foundation
 
 struct PhoenixInternalEvent: PhoenixEvent {
   let kind: PhoenixInternalEventKind
-  let status: Status
+  let status: Status?
 }
 
 // MARK: PhoenixEventCodable
 extension PhoenixInternalEvent: PhoenixEventCodable {
   init(from decoder: Decoder, kind: PhoenixInternalEventKind) throws {
     let container = try decoder.container(keyedBy: CodingKeys.self)
-    let status = try container.decode(Status.self, forKey: .status)
+    let status = try container.decodeIfPresent(Status.self, forKey: .status)
     self.init(kind: kind, status: status)
   }
 

Sources/OpenSeaSwift/Internal/Phoenix/PhoenixInternalEventKind.swift

@@ -8,4 +8,5 @@ enum PhoenixInternalEventKind: String, Codable {
   case heartbeat = "heartbeat"
   case reply = "phx_reply"
   case leave = "phx_leave"
+  case close = "phx_close"
 }

Sources/OpenSeaSwift/Internal/Phoenix/PhoenixReplySubscriptionRef.swift

@@ -4,7 +4,8 @@
 import Foundation
 
 struct PhoenixReplySubscriptionRef: Hashable {
-  let joinRef: String?
+  // We do not support joining the same topic multiple times
+  // let joinRef: String?
   let ref: String?
   let topic: String
 }

Sources/OpenSeaSwift/Internal/Phoenix/PhoenixSocketService.swift

@@ -89,6 +89,8 @@ private extension PhoenixSocketServiceImpl {
   func clear() {
     heartbeatTask?.cancel()
     heartbeatTask = nil
+    topicSubscriptions.removeAll()
+    replySubscriptions.removeAll()
   }
 
   func handle(rawMessage: String) {
@@ -102,7 +104,7 @@ private extension PhoenixSocketServiceImpl {
         try handleMessage(messageData: data)
       }
     } catch {
-      print("Invalid message: \(rawMessage)")
+      print("Invalid message: \(rawMessage) (\(error))")
     }
   }
 
@@ -113,6 +115,8 @@ private extension PhoenixSocketServiceImpl {
       let ref = replySubscriptionRef(forMessage: phoenixMessage)
       guard let subscription = replySubscription(forRef: ref) else { return /* No subscription */ }
       subscription.resume(message: phoenixMessage)
+    case .close:
+      clearTopicSubscription(topic: phoenixMessage.topic)
     default:
       break
     }
@@ -140,8 +144,8 @@ private extension PhoenixSocketServiceImpl {
       try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, Error>) in
         subscription.sink { message in
           switch message.event.status {
-          case .ok:     continuation.resume()
-          case .error:  continuation.resume(throwing: PhoenixSocketError.replyError)
+          case .ok, .none:  continuation.resume()
+          case .error:      continuation.resume(throwing: PhoenixSocketError.replyError)
           }
         }
       }
@@ -156,7 +160,7 @@ private extension PhoenixSocketServiceImpl {
 // MARK: Subscription
 private extension PhoenixSocketServiceImpl {
   func replySubscriptionRef<Event>(forMessage message: PhoenixMessage<Event>) -> PhoenixReplySubscriptionRef {
-    .init(joinRef: message.joinRef, ref: message.ref, topic: message.topic)
+    .init(ref: message.ref, topic: message.topic)
   }
 
   func replySubscription(forRef ref: PhoenixReplySubscriptionRef) -> PhoenixReplySubscription? {

Sources/OpenSeaSwift/Stream/Events/StreamEventPayload.swift

@@ -27,7 +27,31 @@ public extension StreamEvent {
   }
 }
 
-protocol StreamEventPayload {
+extension StreamEvent.Payload: StreamEventPayload {
+  public var item: StreamItem { subPayload.item }
+  public var collection: StreamCollection { subPayload.collection }
+  
+  private var subPayload: StreamEventPayload {
+    switch self {
+    case let .itemMetadataUpdated(itemMetadataUpdated):
+      return itemMetadataUpdated
+    case let .itemListed(itemListed):
+      return itemListed
+    case let .itemSold(itemSold):
+      return itemSold
+    case let .itemTransferred(itemTransferred):
+      return itemTransferred
+    case let .itemCancelled(itemCancelled):
+      return itemCancelled
+    case let .itemReceivedOffer(itemReceivedOffer):
+      return itemReceivedOffer
+    case let .itemReceivedBid(itemReceivedBid):
+      return itemReceivedBid
+    }
+  }
+}
+
+public protocol StreamEventPayload {
   var item: StreamItem { get }
   var collection: StreamCollection { get }
 }