This repository has been archived by the owner on Dec 19, 2021. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 24
/
Copy pathCRVStompClient.m
327 lines (276 loc) · 10.8 KB
/
CRVStompClient.m
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
//
// CRVStompClient.h
// Objc-Stomp
//
//
// Implements the Stomp Protocol v1.0
// See: http://stomp.codehaus.org/Protocol
//
// Requires the AsyncSocket library
// See: http://code.google.com/p/cocoaasyncsocket/
//
// See: LICENSE
// Stefan Saasen <[email protected]>
// Based on StompService.{h,m} by Scott Raymond <[email protected]>.
#import "CRVStompClient.h"
#define kStompDefaultPort 61613
#define kDefaultTimeout 5 //
// ============= http://stomp.codehaus.org/Protocol =============
#define kCommandConnect @"CONNECT"
#define kCommandSend @"SEND"
#define kCommandSubscribe @"SUBSCRIBE"
#define kCommandUnsubscribe @"UNSUBSCRIBE"
#define kCommandBegin @"BEGIN"
#define kCommandCommit @"COMMIT"
#define kCommandAbort @"ABORT"
#define kCommandAck @"ACK"
#define kCommandDisconnect @"DISCONNECT"
#define kControlChar [NSString stringWithFormat:@"\n%C", 0] // TODO -> static
#define kAckClient @"client"
#define kAckAuto @"auto"
#define kResponseHeaderSession @"session"
#define kResponseHeaderReceiptId @"receipt-id"
#define kResponseHeaderErrorMessage @"message"
#define kResponseFrameConnected @"CONNECTED"
#define kResponseFrameMessage @"MESSAGE"
#define kResponseFrameReceipt @"RECEIPT"
#define kResponseFrameError @"ERROR"
// ============= http://stomp.codehaus.org/Protocol =============
#define CRV_RELEASE_SAFELY(__POINTER) { [__POINTER release]; __POINTER = nil; }
@interface CRVStompClient()
@property (nonatomic, assign) NSUInteger port;
@property (nonatomic, retain) AsyncSocket *socket;
@property (nonatomic, copy) NSString *host;
@property (nonatomic, copy) NSString *login;
@property (nonatomic, copy) NSString *passcode;
@property (nonatomic, copy) NSString *sessionId;
@end
@interface CRVStompClient(PrivateMethods)
- (void) sendFrame:(NSString *) command withHeader:(NSDictionary *) header andBody:(NSString *) body;
- (void) sendFrame:(NSString *) command;
- (void) readFrame;
@end
@implementation CRVStompClient
@synthesize delegate;
@synthesize socket, host, port, login, passcode, sessionId;
- (id)init {
return [self initWithHost:@"localhost" port:kStompDefaultPort login:nil passcode:nil delegate:nil];
}
- (id)initWithHost:(NSString *)theHost
port:(NSUInteger)thePort
delegate:(id<CRVStompClientDelegate>)theDelegate
autoconnect:(BOOL) autoconnect {
if(self = [self initWithHost:theHost port:thePort login:nil passcode:nil delegate:theDelegate autoconnect: NO]) {
anonymous = YES;
}
return self;
}
- (id)initWithHost:(NSString *)theHost
port:(NSUInteger)thePort
login:(NSString *)theLogin
passcode:(NSString *)thePasscode
delegate:(id<CRVStompClientDelegate>)theDelegate {
return [self initWithHost:theHost port:thePort login:theLogin passcode:thePasscode delegate:theDelegate autoconnect: NO];
}
- (id)initWithHost:(NSString *)theHost
port:(NSUInteger)thePort
login:(NSString *)theLogin
passcode:(NSString *)thePasscode
delegate:(id<CRVStompClientDelegate>)theDelegate
autoconnect:(BOOL) autoconnect {
if(self = [super init]) {
anonymous = NO;
doAutoconnect = autoconnect;
AsyncSocket *theSocket = [[AsyncSocket alloc] initWithDelegate:self];
[self setSocket: theSocket];
[theSocket release];
[self setDelegate:theDelegate];
[self setHost: theHost];
[self setPort: thePort];
[self setLogin: theLogin];
[self setPasscode: thePasscode];
NSError *err;
if(![self.socket connectToHost:self.host onPort:self.port error:&err]) {
NSLog(@"StompService error: %@", err);
}
}
return self;
}
#pragma mark -
#pragma mark Public methods
- (void)connect {
if(anonymous) {
[self sendFrame:kCommandConnect];
} else {
NSDictionary *headers = [NSDictionary dictionaryWithObjectsAndKeys: [self login], @"login", [self passcode], @"passcode", nil];
[self sendFrame:kCommandConnect withHeader:headers andBody: nil];
}
[self readFrame];
}
- (void)sendMessage:(NSString *)theMessage toDestination:(NSString *)destination {
[self sendMessage:theMessage toDestination:destination withHeaders:[NSDictionary dictionary]];
}
- (void)sendMessage:(NSString *)theMessage toDestination:(NSString *)destination withHeaders:(NSDictionary*)headers {
NSMutableDictionary *allHeaders = [NSMutableDictionary dictionaryWithDictionary:headers];
[allHeaders setValue:destination forKey:@"destination"];
[self sendFrame:kCommandSend withHeader:allHeaders andBody:theMessage];
}
- (void)subscribeToDestination:(NSString *)destination {
[self subscribeToDestination:destination withAck: CRVStompAckModeAuto];
}
- (void)subscribeToDestination:(NSString *)destination withAck:(CRVStompAckMode) ackMode {
NSString *ack;
switch (ackMode) {
case CRVStompAckModeClient:
ack = kAckClient;
break;
default:
ack = kAckAuto;
break;
}
NSDictionary *headers = [NSDictionary dictionaryWithObjectsAndKeys: destination, @"destination", ack, @"ack", nil];
[self sendFrame:kCommandSubscribe withHeader:headers andBody:nil];
}
- (void)subscribeToDestination:(NSString *)destination withHeader:(NSDictionary *) header {
NSMutableDictionary *headers = [[NSMutableDictionary alloc] initWithDictionary:header];
[headers setObject:destination forKey:@"destination"];
[self sendFrame:kCommandSubscribe withHeader:headers andBody:nil];
[headers release];
}
- (void)unsubscribeFromDestination:(NSString *)destination {
NSDictionary *headers = [NSDictionary dictionaryWithObjectsAndKeys: destination, @"destination", nil];
[self sendFrame:kCommandUnsubscribe withHeader:headers andBody:nil];
}
-(void)begin:(NSString *)transactionId {
NSDictionary *headers = [NSDictionary dictionaryWithObjectsAndKeys: transactionId, @"transaction", nil];
[self sendFrame:kCommandBegin withHeader:headers andBody:nil];
}
- (void)commit:(NSString *)transactionId {
NSDictionary *headers = [NSDictionary dictionaryWithObjectsAndKeys: transactionId, @"transaction", nil];
[self sendFrame:kCommandCommit withHeader:headers andBody:nil];
}
- (void)abort:(NSString *)transactionId {
NSDictionary *headers = [NSDictionary dictionaryWithObjectsAndKeys: transactionId, @"transaction", nil];
[self sendFrame:kCommandAbort withHeader:headers andBody:nil];
}
- (void)ack:(NSString *)messageId {
NSDictionary *headers = [NSDictionary dictionaryWithObjectsAndKeys: messageId, @"message-id", nil];
[self sendFrame:kCommandAck withHeader:headers andBody:nil];
}
- (void)disconnect {
[self sendFrame:kCommandDisconnect];
[[self socket] disconnectAfterReadingAndWriting];
}
#pragma mark -
#pragma mark PrivateMethods
- (void) sendFrame:(NSString *) command withHeader:(NSDictionary *) header andBody:(NSString *) body {
NSMutableString *frameString = [NSMutableString stringWithString: [command stringByAppendingString:@"\n"]];
for (id key in header) {
[frameString appendString:key];
[frameString appendString:@":"];
[frameString appendString:[header objectForKey:key]];
[frameString appendString:@"\n"];
}
if (body) {
[frameString appendString:@"\n"];
[frameString appendString:body];
}
[frameString appendString:kControlChar];
[[self socket] writeData:[frameString dataUsingEncoding:NSUTF8StringEncoding] withTimeout:kDefaultTimeout tag:123];
}
- (void) sendFrame:(NSString *) command {
[self sendFrame:command withHeader:nil andBody:nil];
}
- (void)receiveFrame:(NSString *)command headers:(NSDictionary *)headers body:(NSString *)body {
//NSLog(@"receiveCommand '%@' [%@], @%", command, headers, body);
// Connected
if([kResponseFrameConnected isEqual:command]) {
if([[self delegate] respondsToSelector:@selector(stompClientDidConnect:)]) {
[[self delegate] stompClientDidConnect:self];
}
// store session-id
NSString *sessId = [headers valueForKey:kResponseHeaderSession];
[self setSessionId: sessId];
// Response
} else if([kResponseFrameMessage isEqual:command]) {
[[self delegate] stompClient:self messageReceived:body withHeader:headers];
// Receipt
} else if([kResponseFrameReceipt isEqual:command]) {
if([[self delegate] respondsToSelector:@selector(serverDidSendReceipt:withReceiptId:)]) {
NSString *receiptId = [headers valueForKey:kResponseHeaderReceiptId];
[[self delegate] serverDidSendReceipt:self withReceiptId: receiptId];
}
// Error
} else if([kResponseFrameError isEqual:command]) {
if([[self delegate] respondsToSelector:@selector(serverDidSendError:withErrorMessage:detailedErrorMessage:)]) {
NSString *msg = [headers valueForKey:kResponseHeaderErrorMessage];
[[self delegate] serverDidSendError:self withErrorMessage: msg detailedErrorMessage: body];
}
}
}
- (void)readFrame {
[[self socket] readDataToData:[AsyncSocket ZeroData] withTimeout:-1 tag:0];
}
#pragma mark -
#pragma mark AsyncSocketDelegate
- (void)onSocket:(AsyncSocket *)sock didReadData:(NSData*)data withTag:(long)tag {
NSData *strData = [data subdataWithRange:NSMakeRange(0, [data length])];
NSString *msg = [[NSString alloc] initWithData:strData encoding:NSUTF8StringEncoding];
NSMutableArray *contents = (NSMutableArray *)[[msg componentsSeparatedByString:@"\n"] mutableCopy];
if([[contents objectAtIndex:0] isEqual:@""]) {
[contents removeObjectAtIndex:0];
}
NSString *command = [[[contents objectAtIndex:0] copy] autorelease];
NSMutableDictionary *headers = [[[NSMutableDictionary alloc] init] autorelease];
NSMutableString *body = [[[NSMutableString alloc] init] autorelease];
BOOL hasHeaders = NO;
[contents removeObjectAtIndex:0];
for(NSString *line in contents) {
if(hasHeaders) {
[body appendString:line];
} else {
if ([line isEqual:@""]) {
hasHeaders = YES;
} else {
// message-id can look like this: message-id:ID:macbook-pro.local-50389-1237007652070-5:6:-1:1:1
NSMutableArray *parts = [NSMutableArray arrayWithArray:[line componentsSeparatedByString:@":"]];
// key ist the first part
NSString *key = [parts objectAtIndex:0];
[parts removeObjectAtIndex:0];
[headers setObject:[parts componentsJoinedByString:@":"] forKey:key];
}
}
}
[msg release];
[self receiveFrame:command headers:headers body:body];
[self readFrame];
[contents release];
}
- (void)onSocket:(AsyncSocket *)sock didConnectToHost:(NSString *)host port:(UInt16)port {
if(doAutoconnect) {
[self connect];
}
}
- (void)onSocket:(AsyncSocket *)sock didWriteDataWithTag:(long)tag {
}
- (void)onSocketDidDisconnect:(AsyncSocket *)sock {
if([[self delegate] respondsToSelector:@selector(stompClientDidDisconnect:)]) {
[[self delegate] stompClientDidDisconnect: self];
}
}
- (void)onSocket:(AsyncSocket *)sock willDisconnectWithError:(NSError *)err {
if([[self delegate] respondsToSelector:@selector(stompClientWillDisconnect:withError:)]) {
[[self delegate] stompClientWillDisconnect:self withError:err];
}
}
#pragma mark -
#pragma mark Memory management
-(void) dealloc {
delegate = nil;
CRV_RELEASE_SAFELY(passcode);
CRV_RELEASE_SAFELY(login);
CRV_RELEASE_SAFELY(host);
CRV_RELEASE_SAFELY(socket);
[super dealloc];
}
@end