Add CRUD operation on streams (#72)

Signed-off-by: Piotr Piotrowski <piotr@synadia.com>
Co-authored-by: Ziya Suzen <ziya@synadia.com>

Sources/JetStream/JetStreamContext+Stream.swift

@@ -0,0 +1,80 @@
+// Copyright 2024 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import Foundation
+
+/// Extension to `JetStreamContext` adding stream management functionalities.
+extension JetStreamContext {
+
+    /// Creates a stream with the specified configuration.
+    /// Throws an error if the stream configuration is invalid or a stream with given name already exists and has different configuration.
+    public func createStream(cfg: StreamConfig) async throws -> Stream {
+        try Stream.validate(name: cfg.name)
+        let req = try! JSONEncoder().encode(cfg)
+        let subj = "STREAM.CREATE.\(cfg.name)"
+        let info: Response<StreamInfo> = try await request(subj, message: req)
+        switch info {
+        case .success(let info):
+            return Stream(ctx: self, info: info)
+        case .error(let apiResponse):
+            throw apiResponse.error
+        }
+    }
+
+    /// Retrieves a stream by its name.
+    /// Throws an error if the stream does not exist.
+    public func getStream(name: String) async throws -> Stream {
+        try Stream.validate(name: name)
+        let subj = "STREAM.INFO.\(name)"
+        let info: Response<StreamInfo> = try await request(subj)
+        switch info {
+        case .success(let info):
+            return Stream(ctx: self, info: info)
+        case .error(let apiResponse):
+            throw apiResponse.error
+        }
+    }
+
+    /// Updates an existing stream with new configuration.
+    /// Throws an error if the stream configuration is invalid or if the stream with provided name does not exist.
+    public func updateStream(cfg: StreamConfig) async throws -> Stream {
+        try Stream.validate(name: cfg.name)
+        let req = try! JSONEncoder().encode(cfg)
+        let subj = "STREAM.UPDATE.\(cfg.name)"
+        let info: Response<StreamInfo> = try await request(subj, message: req)
+        switch info {
+        case .success(let info):
+            return Stream(ctx: self, info: info)
+        case .error(let apiResponse):
+            throw apiResponse.error
+        }
+    }
+
+    /// Deletes a stream by its name.
+    /// Throws an error if the stream does not exist.
+    public func deleteStream(name: String) async throws {
+        try Stream.validate(name: name)
+        let subj = "STREAM.DELETE.\(name)"
+        let info: Response<StreamDeleteResponse> = try await request(subj)
+        switch info {
+        case .success(_):
+            return
+        case .error(let apiResponse):
+            throw apiResponse.error
+        }
+    }
+
+    struct StreamDeleteResponse: Codable {
+        let success: Bool
+    }
+}

Sources/JetStream/context.swift → Sources/JetStream/JetStreamContext.swift

@@ -53,10 +53,12 @@ extension JetStreamContext {
         return AckFuture(sub: sub, timeout: self.timeout)
     }
 
-    internal func request<T: Codable>(_ subject: String, message: Data) async throws -> Response<T>
-    {
+    internal func request<T: Codable>(
+        _ subject: String, message: Data? = nil
+    ) async throws -> Response<T> {
+        let data = message ?? Data()
         let response = try await self.client.request(
-            message, subject: "\(self.prefix).\(subject)", timeout: self.timeout)
+            data, subject: "\(self.prefix).\(subject)", timeout: self.timeout)
 
         let decoder = JSONDecoder()
         guard let payload = response.payload else {
@@ -87,11 +89,14 @@ public struct AckFuture {
                 for try await result in group {
                     // if the result is not empty, return it (or throw status error)
                     if let msg = result {
+                        group.cancelAll()
                         if let status = msg.status, status == StatusCode.noResponders {
                             throw NatsRequestError.noResponders
                         }
                         return msg
                     } else {
+                        group.cancelAll()
+                        try await sub.unsubscribe()
                         // if result is empty, time out
                         throw NatsRequestError.timeout
                     }

Sources/JetStream/errors.swift → Sources/JetStream/JetStreamError.swift

@@ -18,7 +18,7 @@ public struct JetStreamAPIResponse: Codable {
     let error: JetStreamError
 }
 
-public struct JetStreamError: Codable {
+public struct JetStreamError: Codable, Error {
     var code: UInt
     //FIXME(jrm): This should be mapped to predefined JetStream errors from the server.
     var errorCode: ErrorCode

Sources/JetStream/NanoTimeInterval.swift

@@ -0,0 +1,39 @@
+// Copyright 2024 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import Foundation
+
+/// `NanoTimeInterval` represents a time interval in nanoseconds, facilitating high precision time measurements.
+public struct NanoTimeInterval: Codable, Equatable {
+    /// The value of the time interval in seconds.
+    var value: TimeInterval
+
+    public init(_ timeInterval: TimeInterval) {
+        self.value = timeInterval
+    }
+
+    /// Initializes a `NanoTimeInterval` from a decoder, assuming the encoded value is in nanoseconds.
+    public init(from decoder: Decoder) throws {
+        let container = try decoder.singleValueContainer()
+        let nanoseconds = try container.decode(Double.self)
+        self.value = nanoseconds / 1_000_000_000.0
+    }
+
+    /// Encodes this `NanoTimeInterval` into a given encoder, converting the time interval from seconds to nanoseconds.
+    /// This method allows `NanoTimeInterval` to be serialized directly into a format that stores time in nanoseconds.
+    public func encode(to encoder: Encoder) throws {
+        var container = encoder.singleValueContainer()
+        let nanoseconds = self.value * 1_000_000_000.0
+        try container.encode(nanoseconds)
+    }
+}