feat: destination plugin

Sources/Amplitude/Plugins/AmplitudeDestinationPlugin.swift

@@ -1,18 +1,54 @@
 //
-//  File.swift
+//  AmplitudeDestinationPlugin.swift
 //
 //
 //  Created by Marvin Liu on 10/27/22.
 //
 
-import Foundation
+public class AmplitudeDestinationPlugin: DestinationPlugin {
+    public var amplitude: Amplitude? = nil
+    public let type: PluginType = .destination
+    private var pipeline: EventPipeline?
 
-public class AmplitudeDestinationPlugin: Plugin {
-    public var type: PluginType = PluginType.destination
+    internal func enqueue(event: BaseEvent?) {
+        if let e = event {
+            if e.isValid() {
+                pipeline?.put(event: e)
+            } else {
+                logger.error(message: "Event is invalid for missing information like userId and deviceId")
+            }
+        }
 
-    public var amplitude: Amplitude?
+    }
+
+    public func track(event: BaseEvent) -> BaseEvent? {
+        enqueue(event: event)
+        return event
+    }
+
+    public func identify(event: IdentifyEvent) -> IdentifyEvent? {
+        enqueue(event: event)
+        return event
+    }
+
+    public func groupIdentify(event: GroupIdentifyEvent) -> GroupIdentifyEvent? {
+        enqueue(event: event)
+        return event
+    }
+
+    public func revenue(event: RevenueEvent) -> RevenueEvent? {
+        enqueue(event: event)
+        return event
+    }
+
+    public func flush() {
+        pipeline?.flush()
+    }
 
     public func setup(amplitude: Amplitude) {
+        self.amplitude = amplitude
+        pipeline = EventPipeline(amplitude: amplitude)
+        pipeline?.start()
     }
 
     public func execute(event: BaseEvent) -> BaseEvent? {

Sources/Amplitude/Plugins/DestinationPlugin.swift

@@ -0,0 +1,61 @@
+//
+//  DestinationPlugin.swift
+//
+//
+//  Created by Hao Yu on 11/15/22.
+//
+
+public protocol DestinationPlugin: EventPlugin {
+
+}
+
+private var _timeline: Timeline? = nil
+extension DestinationPlugin {
+    var timeline: Timeline? {
+        if _timeline == nil {
+            _timeline = Timeline()
+        }
+        return _timeline
+    }
+    var enabled: Bool {
+        return true
+    }
+
+    var logger: (any Logger) {
+        return self.amplitude?.logger ?? ConsoleLogger()
+    }
+
+    @discardableResult
+    func add(plugin: Plugin) -> Plugin {
+        plugin.amplitude = self.amplitude
+        timeline?.add(plugin: plugin)
+        return plugin
+    }
+
+    func remove(plugin: Plugin) {
+        timeline?.remove(plugin: plugin)
+    }
+
+    func process(event: BaseEvent?) -> BaseEvent? {
+        // Skip this destination if it is disabled via settings
+        if !enabled {
+            return nil
+        }
+        let beforeResult = timeline?.applyPlugin(pluginType: .before, event: event)
+        let enrichmentResult = timeline?.applyPlugin(pluginType: .enrichment, event: beforeResult)
+        var destinationResult: BaseEvent? = nil
+        switch enrichmentResult {
+        case let e as IdentifyEvent:
+            destinationResult = identify(event: e)
+        case let e as GroupIdentifyEvent:
+            destinationResult = track(event: e)
+        case let e as RevenueEvent:
+            destinationResult = revenue(event: e)
+        case let e?:
+            destinationResult = track(event: e)
+        default:
+            break
+        }
+        return destinationResult
+    }
+}

Sources/Amplitude/Types.swift

@@ -58,6 +58,3 @@ public protocol EventPlugin: Plugin {
     func revenue(event: RevenueEvent) -> RevenueEvent?
     func flush()
 }
-
-public protocol DestinationPlugin: EventPlugin {
-}