swift-SWIM

dependencies: [
    .package(url: "https://github.com/1amageek/swift-SWIM.git", from: "1.0.0")
]

.target(
    name: "YourTarget",
    dependencies: ["SWIM"]
)

import SWIM

// 1. Create a transport (implement SWIMTransport protocol)
let transport = MyUDPTransport(localAddress: "192.168.1.1:8000")

// 2. Create a SWIM instance
let localMember = Member(id: MemberID(id: "node1", address: "192.168.1.1:8000"))
let swim = SWIMInstance(
    localMember: localMember,
    config: .default,
    transport: transport
)

// 3. Start and join the cluster
await swim.start()
try await swim.join(seeds: [seedMemberID])

// 4. Monitor membership changes
for await event in swim.events {
    switch event {
    case .memberJoined(let member):
        print("Joined: \(member)")
    case .memberSuspected(let member):
        print("Suspected: \(member)")
    case .memberFailed(let member):
        print("Failed: \(member)")
    case .memberRecovered(let member):
        print("Recovered: \(member)")
    case .memberLeft(let id):
        print("Left: \(id)")
    default:
        break
    }
}

public protocol SWIMTransport: Sendable {
    /// Send a message
    func send(_ message: SWIMMessage, to member: MemberID) async throws

    /// Stream of incoming messages
    var incomingMessages: AsyncStream<(SWIMMessage, MemberID)> { get }

    /// Local address
    var localAddress: String { get }
}

final class UDPTransport: SWIMTransport, Sendable {
    let localAddress: String
    let incomingMessages: AsyncStream<(SWIMMessage, MemberID)>
    private let continuation: AsyncStream<(SWIMMessage, MemberID)>.Continuation
    private let socket: UDPSocket

    init(localAddress: String) async throws {
        self.localAddress = localAddress

        var cont: AsyncStream<(SWIMMessage, MemberID)>.Continuation!
        self.incomingMessages = AsyncStream { cont = $0 }
        self.continuation = cont

        self.socket = try await UDPSocket(address: localAddress)

        // Start receiving
        Task {
            for await (data, address) in socket.incoming {
                if let message = try? SWIMMessageCodec.decode(data) {
                    let sender = MemberID(id: address, address: address)
                    continuation.yield((message, sender))
                }
            }
        }
    }

    func send(_ message: SWIMMessage, to member: MemberID) async throws {
        let data = SWIMMessageCodec.encode(message)
        try await socket.send(data, to: member.address)
    }
}

var config = SWIMConfiguration()

// Protocol period (time between probes)
config.protocolPeriod = .milliseconds(200)

// Ping response timeout
config.pingTimeout = .milliseconds(100)

// Number of indirect probes
config.indirectProbeCount = 3

// Suspicion timeout multiplier
config.suspicionMultiplier = 5.0

// Maximum updates per message
config.maxPayloadSize = 10

// Base dissemination count (actual = base * log(N))
config.baseDisseminationLimit = 3

// MockTransport for unit tests
let transport = MockTransport(localAddress: "127.0.0.1:8000")

// Simulate receiving a message
transport.receive(message, from: sender)

// Check sent messages
let sent = transport.getSentMessages()

// LoopbackTransport for integration tests
let transport1 = LoopbackTransport(localAddress: "127.0.0.1:8000")
let transport2 = LoopbackTransport(localAddress: "127.0.0.1:8001")
transport1.connect(to: transport2)
transport2.connect(to: transport1)