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
135
136
137
138
139
140
141
142
143
144
145
146
|
/*
* Copyright 2017 Google
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#import "FIRMessagingDelayedMessageQueue.h"
#import "Protos/GtalkCore.pbobjc.h"
#import "FIRMessagingDefines.h"
#import "FIRMessagingRmqManager.h"
#import "FIRMessagingUtilities.h"
static const int kMaxQueuedMessageCount = 10;
@interface FIRMessagingDelayedMessageQueue ()
@property(nonatomic, readonly, weak) id<FIRMessagingRmqScanner> rmqScanner;
@property(nonatomic, readonly, copy) FIRMessagingSendDelayedMessagesHandler sendDelayedMessagesHandler;
@property(nonatomic, readwrite, assign) int persistedMessageCount;
// the scheduled timeout or -1 if not set
@property(nonatomic, readwrite, assign) int64_t scheduledTimeoutMilliseconds;
// The time of the last scan of the message DB,
// used to avoid retrieving messages more than once.
@property(nonatomic, readwrite, assign) int64_t lastDBScanTimestampSeconds;
@property(nonatomic, readwrite, strong) NSMutableArray *messages;
@property(nonatomic, readwrite, strong) NSTimer *sendTimer;
@end
@implementation FIRMessagingDelayedMessageQueue
- (instancetype)init {
FIRMessagingInvalidateInitializer();
}
- (instancetype)initWithRmqScanner:(id<FIRMessagingRmqScanner>)rmqScanner
sendDelayedMessagesHandler:(FIRMessagingSendDelayedMessagesHandler)sendDelayedMessagesHandler {
_FIRMessagingDevAssert(sendDelayedMessagesHandler, @"Invalid nil callback for delayed messages");
self = [super init];
if (self) {
_rmqScanner = rmqScanner;
_sendDelayedMessagesHandler = sendDelayedMessagesHandler;
_messages = [NSMutableArray arrayWithCapacity:10];
_scheduledTimeoutMilliseconds = -1;
}
return self;
}
- (BOOL)queueMessage:(GtalkDataMessageStanza *)message {
if (self.messages.count >= kMaxQueuedMessageCount) {
return NO;
}
if (message.ttl == 0) {
// ttl=0 messages aren't persisted, add it to memory
[self.messages addObject:message];
} else {
self.persistedMessageCount++;
}
int64_t timeoutMillis = [self calculateTimeoutInMillisWithDelayInSeconds:message.maxDelay];
if (![self isTimeoutScheduled] || timeoutMillis < self.scheduledTimeoutMilliseconds) {
[self scheduleTimeoutInMillis:timeoutMillis];
}
return YES;
}
- (NSArray *)removeDelayedMessages {
[self cancelTimeout];
if ([self messageCount] == 0) {
return @[];
}
NSMutableArray *delayedMessages = [NSMutableArray array];
// add the ttl=0 messages
if (self.messages.count) {
[delayedMessages addObjectsFromArray:delayedMessages];
[self.messages removeAllObjects];
}
// add persistent messages
if (self.persistedMessageCount > 0) {
FIRMessaging_WEAKIFY(self);
[self.rmqScanner scanWithRmqMessageHandler:nil
dataMessageHandler:^(int64_t rmqId, GtalkDataMessageStanza *stanza) {
FIRMessaging_STRONGIFY(self);
if ([stanza hasMaxDelay] &&
[stanza sent] >= self.lastDBScanTimestampSeconds) {
[delayedMessages addObject:stanza];
}
}];
self.lastDBScanTimestampSeconds = FIRMessagingCurrentTimestampInSeconds();
self.persistedMessageCount = 0;
}
return delayedMessages;
}
- (void)sendMessages {
if (self.sendDelayedMessagesHandler) {
self.sendDelayedMessagesHandler([self removeDelayedMessages]);
}
}
#pragma mark - Private
- (NSInteger)messageCount {
return self.messages.count + self.persistedMessageCount;
}
- (BOOL)isTimeoutScheduled {
return self.scheduledTimeoutMilliseconds > 0;
}
- (int64_t)calculateTimeoutInMillisWithDelayInSeconds:(int)delay {
return FIRMessagingCurrentTimestampInMilliseconds() + delay * 1000.0;
}
- (void)scheduleTimeoutInMillis:(int64_t)time {
[self cancelTimeout];
self.scheduledTimeoutMilliseconds = time;
double delay = (time - FIRMessagingCurrentTimestampInMilliseconds()) / 1000.0;
[self performSelector:@selector(sendMessages) withObject:self afterDelay:delay];
}
- (void)cancelTimeout {
if ([self isTimeoutScheduled]) {
[NSObject cancelPreviousPerformRequestsWithTarget:self
selector:@selector(sendMessages)
object:nil];
self.scheduledTimeoutMilliseconds = -1;
}
}
@end
|
