-
Notifications
You must be signed in to change notification settings - Fork 38
/
AMQPQueue.m
75 lines (59 loc) · 2.8 KB
/
AMQPQueue.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
//
// AMQPQueue.m
// This file is part of librabbitmq-objc.
// Copyright (C) 2014 *Prof. MAAD* aka Max Wolter
// librabbitmq-objc is released under the terms of the GNU Lesser General Public License Version 3.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
#import "AMQPQueue.h"
# import <amqp.h>
# import <amqp_framing.h>
# import "AMQPChannel.h"
# import "AMQPExchange.h"
# import "AMQPConsumer.h"
@implementation AMQPQueue
@synthesize internalQueue = queueName;
- (id)initWithName:(NSString*)theName onChannel:(AMQPChannel*)theChannel isPassive:(BOOL)passive isExclusive:(BOOL)exclusive isDurable:(BOOL)durable getsAutoDeleted:(BOOL)autoDelete
{
if(self = [super init])
{
amqp_queue_declare_ok_t *declaration = amqp_queue_declare(theChannel.connection.internalConnection, theChannel.internalChannel, amqp_cstring_bytes([theName UTF8String]), passive, durable, exclusive, autoDelete, AMQP_EMPTY_TABLE);
[theChannel.connection checkLastOperation:@"Failed to declare queue"];
queueName = amqp_bytes_malloc_dup(declaration->queue);
channel = [theChannel retain];
}
return self;
}
- (void)dealloc
{
amqp_bytes_free(queueName);
[channel release];
[super dealloc];
}
- (void)bindToExchange:(AMQPExchange*)theExchange withKey:(NSString*)bindingKey
{
amqp_queue_bind(channel.connection.internalConnection, channel.internalChannel, queueName, theExchange.internalExchange, amqp_cstring_bytes([bindingKey UTF8String]), AMQP_EMPTY_TABLE);
[channel.connection checkLastOperation:@"Failed to bind queue to exchange"];
}
- (void)unbindFromExchange:(AMQPExchange*)theExchange withKey:(NSString*)bindingKey
{
amqp_queue_unbind(channel.connection.internalConnection, channel.internalChannel, queueName, theExchange.internalExchange, amqp_cstring_bytes([bindingKey UTF8String]), AMQP_EMPTY_TABLE);
[channel.connection checkLastOperation:@"Failed to unbind queue from exchange"];
}
- (AMQPConsumer*)startConsumerWithAcknowledgements:(BOOL)ack isExclusive:(BOOL)exclusive receiveLocalMessages:(BOOL)local
{
AMQPConsumer *consumer = [[AMQPConsumer alloc] initForQueue:self onChannel:channel useAcknowledgements:ack isExclusive:exclusive receiveLocalMessages:local];
return [consumer autorelease];
}
@end