-
Notifications
You must be signed in to change notification settings - Fork 32
Expand file tree
/
Copy pathTwift+Streams.swift
More file actions
134 lines (109 loc) 路 7.25 KB
/
Copy pathTwift+Streams.swift
File metadata and controls
134 lines (109 loc) 路 7.25 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
import Foundation
extension Twift {
// MARK: Streams
/// Streams about 1% of all Tweets in real-time.
///
/// Equivalent to `GET /2/tweets/sample/stream`
/// - Parameters:
/// - fields: Any additional fields to include on returned objects
/// - expansions: Objects and their corresponding fields that should be expanded in the `includes` property
/// - backfillMinutes: By passing this parameter, you can request up to five (5) minutes worth of streaming data that you might have missed during a disconnection to be delivered to you upon reconnection. The backfilled Tweets will automatically flow through the reconnected stream, with older Tweets generally being delivered before any newly matching Tweets. You must include a whole number between 1 and 5 as the value to this parameter.
/// This feature will deliver duplicate Tweets, meaning that if you were disconnected for 90 seconds, and you requested two minutes of backfill, you will receive 30 seconds worth of duplicate Tweets. Due to this, you should make sure your system is tolerant of duplicate data.
/// This feature is currently only available to the Academic Research product track.
/// - Returns: An `AsyncSequence` of `TwitterAPIDataAndIncludes<Tweet, Tweet.Includes>` objects.
public func volumeStream(fields: Set<Tweet.Field> = [],
expansions: [Tweet.Expansions] = [],
backfillMinutes: Int? = nil
) async throws -> AsyncThrowingCompactMapSequence<AsyncLineSequence<URLSession.AsyncBytes>, TwitterAPIDataAndIncludes<Tweet, Tweet.Includes>> {
guard case .appOnly(_) = authenticationType else { throw TwiftError.WrongAuthenticationType(needs: .appOnly) }
var queryItems = fieldsAndExpansions(for: Tweet.self, fields: fields, expansions: expansions)
if let backfillMinutes = backfillMinutes {
queryItems.append(URLQueryItem(name: "backfill_minutes", value: "\(backfillMinutes)"))
}
let url = getURL(for: .volumeStream, queryItems: queryItems)
var request = URLRequest(url: url)
signURLRequest(method: .GET, request: &request)
let (bytes, response) = try await URLSession.shared.bytes(for: request)
guard let response = response as? HTTPURLResponse,
response.statusCode == 200 else {
throw URLError.init(.resourceUnavailable)
}
return bytes.lines
.compactMap {
try? await self.decodeOrThrow(decodingType: TwitterAPIDataAndIncludes.self, data: Data($0.utf8))
}
}
/// Streams Tweets in real-time based on a specific set of filter rules.
///
/// Equivalent to `GET /2/tweets/search/stream`
/// - Parameters:
/// - fields: Any additional fields to include on returned objects
/// - expansions: Objects and their corresponding fields that should be expanded in the `includes` property
/// - backfillMinutes: By passing this parameter, you can request up to five (5) minutes worth of streaming data that you might have missed during a disconnection to be delivered to you upon reconnection. The backfilled Tweets will automatically flow through the reconnected stream, with older Tweets generally being delivered before any newly matching Tweets. You must include a whole number between 1 and 5 as the value to this parameter.
/// This feature will deliver duplicate Tweets, meaning that if you were disconnected for 90 seconds, and you requested two minutes of backfill, you will receive 30 seconds worth of duplicate Tweets. Due to this, you should make sure your system is tolerant of duplicate data.
/// This feature is currently only available to the Academic Research product track.
/// - Returns: An `AsyncSequence` of `TwitterAPIDataAndIncludes<Tweet, Tweet.Includes>` objects.
public func filteredStream(fields: Set<Tweet.Field> = [],
expansions: [Tweet.Expansions] = [],
backfillMinutes: Int? = nil
) async throws -> AsyncThrowingCompactMapSequence<AsyncLineSequence<URLSession.AsyncBytes>, TwitterAPIDataAndIncludes<Tweet, Tweet.Includes>> {
guard case .appOnly(_) = authenticationType else { throw TwiftError.WrongAuthenticationType(needs: .appOnly) }
var queryItems = fieldsAndExpansions(for: Tweet.self, fields: fields, expansions: expansions)
if let backfillMinutes = backfillMinutes {
queryItems.append(URLQueryItem(name: "backfill_minutes", value: "\(backfillMinutes)"))
}
let url = getURL(for: .filteredStream, queryItems: queryItems)
var request = URLRequest(url: url)
signURLRequest(method: .GET, request: &request)
let (bytes, response) = try await URLSession.shared.bytes(for: request)
guard let response = response as? HTTPURLResponse,
response.statusCode == 200 else {
throw URLError.init(.resourceUnavailable)
}
return bytes.lines
.compactMap {
try? await self.decodeOrThrow(decodingType: TwitterAPIDataAndIncludes.self, data: Data($0.utf8))
}
}
}
extension Twift {
// MARK: Stream Rules
/// Return a list of rules currently active on the streaming endpoint, either as a list or individually.
///
/// Equivalent to `get /2/tweets/search/stream/rules`
/// - Parameter ids: A list of rule IDs to return. If omitted, all rules are returned.
/// - Returns: A response object containing an array of fetched stream rules.
public func getFilteredStreamRules(ids: [FilteredStreamRule.ID]? = nil)
async throws -> TwitterAPIDataAndMeta<[FilteredStreamRule], FilteredStreamRuleMeta> {
var queryItems: [URLQueryItem] = []
if let ids = ids, !ids.isEmpty {
queryItems.append(URLQueryItem(name: "ids", value: ids.joined(separator: ",")))
}
return try await call(route: .filteredStreamRules,
queryItems: queryItems)
}
/// Add or delete rules to your stream.
///
/// Equivalent to `POST /2/tweets/search/stream/rules`
/// - Parameters:
/// - add: An array of rule objects to add to the filtered stream
/// - delete: An array of rule IDs to delete from the filtered stream
/// - dryRun: Set to true to test a the syntax of your rule without submitting it. This is useful if you want to check the syntax of a rule before removing one or more of your existing rules.
/// - Returns: A response object containing an optional array of rules created by the request and a meta object with details of any rules created/deleted by the request
public func modifyFilteredStreamRules(add: [MutableFilteredStreamRule] = [],
delete: [FilteredStreamRule.ID] = [],
dryRun: Bool = false
) async throws -> TwitterAPIDataAndMeta<[FilteredStreamRule], FilteredStreamRuleMeta> {
var queryItems: [URLQueryItem] = []
if dryRun { queryItems.append(URLQueryItem(name: "dry_run", value: "true")) }
let modifier = FilteredStreamRuleModifier(add: add, delete: delete)
let serializedBody = try self.encoder.encode(modifier)
return try await call(route: .filteredStreamRules,
method: .POST,
body: serializedBody)
}
}
internal struct FilteredStreamRuleModifier: Codable {
var add: [MutableFilteredStreamRule] = []
var delete: [FilteredStreamRule.ID] = []
}