Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Make Segment.io flush asynchronous #259

Closed
wants to merge 5 commits into from
Closed

Make Segment.io flush asynchronous #259

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
3066c42
Fixed JSON conversion test for swift 5.9
bsneed Oct 13, 2023
4c3311d
Make segment.io flush asynchronous.
bsneed Oct 16, 2023
efd762c
Merge remote-tracking branch 'origin/main' into bsneed/async-flush
bsneed Oct 16, 2023
e3c79ba
Fix self syntax differences between swift 5.8 and 5.9.
bsneed Oct 16, 2023
7745104
missed one.
bsneed Oct 16, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 38 additions & 34 deletions Sources/Segment/Plugins/SegmentDestination.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,8 @@ public class SegmentDestination: DestinationPlugin, Subscriber {

internal var httpClient: HTTPClient?
private var uploads = [UploadTaskInfo]()
private let uploadsQueue = DispatchQueue(label: "uploadsQueue.segment.com")
private let uploadsQueue = DispatchQueue(label: "com.segment.uploadsQueue")
private let flushQueue = DispatchQueue(label: "com.segment.flushQueue")
private var storage: Storage?

@Atomic internal var eventCount: Int = 0
Expand Down Expand Up @@ -113,47 +114,50 @@ public class SegmentDestination: DestinationPlugin, Subscriber {
}

public func flush() {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems like this could be simplified to:

public func flush() {
    flushQueue.async { [weak self] in
        self?._flush()
    }
}

// Existing flush (no need for this comment-- just making the point)
private func _flush() {
  // existing code...
}

guard let storage = self.storage else { return }
guard let analytics = self.analytics else { return }
guard let httpClient = self.httpClient else { return }

// don't flush if analytics is disabled.
guard analytics.enabled == true else { return }

// Read events from file system
guard let data = storage.read(Storage.Constants.events) else { return }

eventCount = 0
cleanupUploads()

analytics.log(message: "Uploads in-progress: \(pendingUploads)")

if pendingUploads == 0 {
for url in data {
analytics.log(message: "Processing Batch:\n\(url.lastPathComponent)")

let uploadTask = httpClient.startBatchUpload(writeKey: analytics.configuration.values.writeKey, batch: url) { (result) in
switch result {
flushQueue.async { [weak self] in
guard let self = self else { return }
guard let storage = self.storage else { return }
guard let analytics = self.analytics else { return }
guard let httpClient = self.httpClient else { return }

// don't flush if analytics is disabled.
guard analytics.enabled == true else { return }

// Read events from file system
guard let data = storage.read(Storage.Constants.events) else { return }

self.eventCount = 0
self.cleanupUploads()

analytics.log(message: "Uploads in-progress: \(self.pendingUploads)")

if self.pendingUploads == 0 {
for url in data {
analytics.log(message: "Processing Batch:\n\(url.lastPathComponent)")

let uploadTask = httpClient.startBatchUpload(writeKey: analytics.configuration.values.writeKey, batch: url) { (result) in
switch result {
case .success(_):
storage.remove(file: url)
self.cleanupUploads()
default:
break
}

analytics.log(message: "Processed: \(url.lastPathComponent)")
// the upload we have here has just finished.
// make sure it gets removed and it's cleanup() called rather
// than waiting on the next flush to come around.
self.cleanupUploads()
}
// we have a legit upload in progress now, so add it to our list.
if let upload = uploadTask {
self.add(uploadTask: UploadTaskInfo(url: url, task: upload))
}

analytics.log(message: "Processed: \(url.lastPathComponent)")
// the upload we have here has just finished.
// make sure it gets removed and it's cleanup() called rather
// than waiting on the next flush to come around.
self.cleanupUploads()
}
// we have a legit upload in progress now, so add it to our list.
if let upload = uploadTask {
add(uploadTask: UploadTaskInfo(url: url, task: upload))
}
} else {
analytics.log(message: "Skipping processing; Uploads in progress.")
}
} else {
analytics.log(message: "Skipping processing; Uploads in progress.")
}
}
}
Expand Down
Loading