d-exclaimation / desolate Goto Github PK

• Updated restriction to be macOS v10.15
• Improvement with callbacks (1.4.0)
• Improvement with @sendable and Sendable object for callbacks (1.4.0)
• Removal of unused code (1.4.0)
• Improvement in Source (1.4.0)
• Improvement in Nozzle termination

Data structure that handle multiple topic stream

• Reservoir, Multiple topic, multiple source / upstream, multiple downstream, single type broadcast stream.
• Confluence , Merge combine stream.
• Cascade, Merge concat stream.

Not sure whether the other two will come just yet or should be suspended to focus on optimizing current code

Description

For some reason, I didn't use the built-in Signal status on each Desolated actors and declare unnecessary additional boolean state that require management myself. I should opt in using Signal as much as possible as it works nicely automatically and does not require manual state management.

Code mentioned

• desolate/Sources/Desolate/Streaming/Jet.swift

  Line 24 in 190f0d4

  private var closed: Bool = false

• desolate/Sources/Desolate/Streaming/Jet.swift

  Line 29 in 190f0d4

  guard !closed else { return same }

• desolate/Sources/Desolate/Streaming/Jet.swift

  Line 37 in 190f0d4

  closed = true

Solution

• Removed NonStop from Jet and use the Signal

// (removed) private var closed: Bool = false

public var status: Signal = .running
private var flows: [UUID: Flow] = [:]

/// Interface for Desolate
public func onMessage(msg: Act) async -> Signal {
    // (removed) guard !closed else { return same }

    switch msg {
    case .next(let message):
        for nozzle in flows.values {
            await nozzle.task(with: message)
        }
    case .complete:
        // (removed) closed = true
        for queue in flows.values {
            await queue.task(with: nil)
        }
        flows.removeAll()
        return .stopped
    case .attach(let id, let nozzle):
        flows[id] = nozzle
    case .detach(id: let id):
        flows.removeValue(forKey: id)
    }
    return .running
}

Descriptions

Removed usage of any fatal error and unsafe mutations to prevent unsuspecting exceptions and behavior

Code references

Fatal Error

• desolate/Sources/Desolate/Decorator/Extensions/Desolate+StateHook.swift

  Line 20 in 1072648

  fatalError(err.localizedDescription)

• desolate/Sources/Desolate/Decorator/Desolated.swift

  Line 31 in 1072648

  fatalError("No value given")

• desolate/Sources/Desolate/Delivery/Receiver.swift

  Line 72 in ec0ca05

  public class Receiver<ReceivedType> {

Unsafe mutation

• desolate/Sources/Desolate/Delivery/Inbox.swift

  Line 10 in ec0ca05

  private var cache: MessageType? = nil

Jet stream is a pretty unique name to call a reactive stream but it's not that descriptive. There are plenty of names that would fit better and explain the use of the data structures.

At the moment, Nozzle delay request to the Current for new data, but the delay is static and doesn't scale with the computing power.