// This source code is dual-licensed under the Mozilla Public License ("MPL"),
|
// version 1.1 and the Apache License ("ASL"), version 2.0.
|
//
|
// The ASL v2.0:
|
//
|
// ---------------------------------------------------------------------------
|
// Copyright 2016 Pivotal Software, Inc.
|
//
|
// 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.
|
// ---------------------------------------------------------------------------
|
//
|
// The MPL v1.1:
|
//
|
// ---------------------------------------------------------------------------
|
// The contents of this file are subject to the Mozilla Public License
|
// Version 1.1 (the "License"); you may not use this file except in
|
// compliance with the License. You may obtain a copy of the License at
|
// https://www.mozilla.org/MPL/
|
//
|
// Software distributed under the License is distributed on an "AS IS"
|
// basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
|
// License for the specific language governing rights and limitations
|
// under the License.
|
//
|
// The Original Code is RabbitMQ
|
//
|
// The Initial Developer of the Original Code is Pivotal Software, Inc.
|
// All Rights Reserved.
|
//
|
// Alternatively, the contents of this file may be used under the terms
|
// of the Apache Standard license (the "ASL License"), in which case the
|
// provisions of the ASL License are applicable instead of those
|
// above. If you wish to allow use of your version of this file only
|
// under the terms of the ASL License and not to allow others to use
|
// your version of this file under the MPL, indicate your decision by
|
// deleting the provisions above and replace them with the notice and
|
// other provisions required by the ASL License. If you do not delete
|
// the provisions above, a recipient may use your version of this file
|
// under either the MPL or the ASL License.
|
// ---------------------------------------------------------------------------
|
|
#import "RMQBasicProperties+MergeDefaults.h"
|
#import "RMQErrors.h"
|
#import "RMQMethods+Convenience.h"
|
#import "RMQAllocatedChannel.h"
|
#import "RMQConnectionDelegate.h"
|
|
@interface RMQAllocatedChannel ()
|
@property (nonatomic, copy, readwrite) NSNumber *channelNumber;
|
@property (nonatomic, readwrite) NSNumber *contentBodySize;
|
@property (nonatomic, readwrite) id <RMQDispatcher> dispatcher;
|
@property (nonatomic, readwrite) NSMutableDictionary *consumers;
|
@property (nonatomic, readwrite) NSMutableDictionary *exchanges;
|
@property (nonatomic, readwrite) NSMutableDictionary *exchangeBindings;
|
@property (nonatomic, readwrite) NSMutableDictionary *queues;
|
@property (nonatomic, readwrite) NSMutableDictionary *queueBindings;
|
@property (nonatomic, readwrite) id<RMQConfirmations> confirmations;
|
@property (nonatomic, readwrite) NSNumber *prefetchCountPerConsumer;
|
@property (nonatomic, readwrite) NSNumber *prefetchCountPerChannel;
|
@property (nonatomic, readwrite) id<RMQConnectionDelegate> delegate;
|
@property (nonatomic, readwrite) id<RMQNameGenerator> nameGenerator;
|
@property (nonatomic, readwrite) id<RMQChannelAllocator> allocator;
|
@end
|
|
@implementation RMQAllocatedChannel
|
|
- (instancetype)init:(NSNumber *)channelNumber
|
contentBodySize:(NSNumber *)contentBodySize
|
dispatcher:(id<RMQDispatcher>)dispatcher
|
nameGenerator:(id<RMQNameGenerator>)nameGenerator
|
allocator:(nonnull id<RMQChannelAllocator>)allocator
|
confirmations:(id<RMQConfirmations>)confirmations {
|
self = [super init];
|
if (self) {
|
self.channelNumber = channelNumber;
|
self.contentBodySize = contentBodySize;
|
self.dispatcher = dispatcher;
|
self.consumers = [NSMutableDictionary new];
|
self.exchanges = [NSMutableDictionary new];
|
self.exchangeBindings = [NSMutableDictionary new];
|
self.queues = [NSMutableDictionary new];
|
self.queueBindings = [NSMutableDictionary new];
|
self.confirmations = confirmations;
|
self.prefetchCountPerConsumer = nil;
|
self.prefetchCountPerChannel = nil;
|
self.delegate = nil;
|
self.nameGenerator = nameGenerator;
|
self.allocator = allocator;
|
}
|
return self;
|
}
|
|
- (instancetype)init
|
{
|
[self doesNotRecognizeSelector:_cmd];
|
return nil;
|
}
|
|
- (RMQExchange *)defaultExchange {
|
return [[RMQExchange alloc] initWithName:@""
|
type:@"direct"
|
options:RMQExchangeDeclareNoOptions
|
channel:self];
|
}
|
|
- (void)activateWithDelegate:(id<RMQConnectionDelegate>)delegate {
|
[self.dispatcher activateWithChannel:self delegate:delegate];
|
self.delegate = delegate;
|
}
|
|
- (void)open {
|
[self.dispatcher sendSyncMethod:[RMQChannelOpen new]];
|
}
|
|
- (void)close {
|
[self.dispatcher sendSyncMethod:[RMQChannelClose new]
|
completionHandler:^(RMQFrameset *frameset) {
|
[self.allocator releaseChannelNumber:self.channelNumber];
|
}];
|
}
|
|
- (void)blockingClose {
|
[self.dispatcher sendSyncMethodBlocking:[RMQChannelClose new]];
|
[self.allocator releaseChannelNumber:self.channelNumber];
|
}
|
|
- (void)prepareForRecovery {
|
[self.dispatcher disable];
|
}
|
|
- (void)recover {
|
[self open];
|
[self recoverPrefetch];
|
[self recoverConfirmations];
|
[self recoverExchanges];
|
[self recoverExchangeBindings];
|
[self recoverQueuesAndTheirBindings];
|
[self recoverConsumers];
|
}
|
|
- (void)blockingWaitOn:(Class)method {
|
[self.dispatcher blockingWaitOn:method];
|
}
|
|
- (void)confirmSelect {
|
[self.confirmations enable];
|
[self.dispatcher sendSyncMethod:[RMQConfirmSelect new]];
|
}
|
|
- (void)afterConfirmed:(NSNumber *)timeout
|
handler:(void (^)(NSSet<NSNumber *> * _Nonnull, NSSet<NSNumber *> * _Nonnull))handler {
|
[self.confirmations addCallbackWithTimeout:timeout
|
callback:handler];
|
}
|
|
- (void)afterConfirmed:(RMQConfirmationCallback)handler {
|
[self afterConfirmed:@30
|
handler:handler];
|
}
|
|
- (RMQQueue *)queue:(NSString *)originalQueueName
|
options:(RMQQueueDeclareOptions)options
|
arguments:(nonnull NSDictionary<NSString *,RMQValue<RMQFieldValue> *> *)arguments {
|
RMQQueue *found = self.queues[originalQueueName];
|
if (found) {
|
return found;
|
} else {
|
return [self memoizedQueueDeclare:originalQueueName options:options arguments:[[RMQTable alloc] init:arguments]];
|
}
|
}
|
|
- (RMQQueue *)queue:(NSString *)originalQueueName
|
options:(RMQQueueDeclareOptions)options {
|
return [self queue:originalQueueName options:options arguments:@{}];
|
}
|
|
- (RMQQueue *)queue:(NSString *)queueName {
|
return [self queue:queueName options:RMQQueueDeclareNoOptions];
|
}
|
|
- (void)queueDelete:(NSString *)queueName
|
options:(RMQQueueDeleteOptions)options {
|
[self.queues removeObjectForKey:queueName];
|
[self.dispatcher sendSyncMethod:[[RMQQueueDelete alloc] initWithQueue:queueName
|
options:options]];
|
}
|
|
- (void)queueBind:(NSString *)queueName
|
exchange:(NSString *)exchangeName
|
routingKey:(nonnull NSString *)routingKey {
|
[self.dispatcher sendSyncMethod:[[RMQQueueBind alloc] initWithQueue:queueName
|
exchange:exchangeName
|
routingKey:routingKey]];
|
[self.queueBindings[queueName] addObject:@{@"exchange": exchangeName,
|
@"routing-key": routingKey}];
|
}
|
|
- (void)queueUnbind:(NSString *)queueName
|
exchange:(NSString *)exchangeName
|
routingKey:(NSString *)routingKey {
|
[self.dispatcher sendSyncMethod:[[RMQQueueUnbind alloc] initWithQueue:queueName
|
exchange:exchangeName
|
routingKey:routingKey]];
|
[self.queueBindings[queueName] removeObject:@{@"exchange": exchangeName,
|
@"routing-key": routingKey}];
|
}
|
|
- (RMQConsumer *)basicConsume:(NSString *)queueName
|
options:(RMQBasicConsumeOptions)options
|
handler:(RMQConsumerDeliveryHandler)handler {
|
RMQConsumer *consumer = [[RMQConsumer alloc] initWithChannel:self
|
queueName:queueName
|
options:options];
|
[consumer onDelivery:handler];
|
[self basicConsume:consumer];
|
return consumer;
|
}
|
|
- (void)basicConsume:(RMQConsumer *)consumer {
|
[self.dispatcher sendSyncMethod:[[RMQBasicConsume alloc] initWithQueue:consumer.queueName
|
consumerTag:consumer.tag
|
options:consumer.options]
|
completionHandler:^(RMQFrameset *frameset) {
|
self.consumers[consumer.tag] = consumer;
|
}];
|
}
|
|
- (NSString *)generateConsumerTag {
|
return [self.nameGenerator generateWithPrefix:@"rmq-objc-client.gen-"];
|
}
|
|
- (void)basicCancel:(NSString *)consumerTag {
|
[self.dispatcher sendSyncMethod:[[RMQBasicCancel alloc] initWithConsumerTag:[[RMQShortstr alloc] init:consumerTag]
|
options:RMQBasicCancelNoOptions]
|
completionHandler:^(RMQFrameset *frameset) {
|
[self.consumers removeObjectForKey:consumerTag];
|
}];
|
}
|
|
- (NSNumber *)basicPublish:(NSData *)body
|
routingKey:(NSString *)routingKey
|
exchange:(NSString *)exchange
|
properties:(NSArray<RMQValue *> *)properties
|
options:(RMQBasicPublishOptions)options {
|
RMQBasicPublish *publish = [[RMQBasicPublish alloc] initWithReserved1:[[RMQShort alloc] init:0]
|
exchange:[[RMQShortstr alloc] init:exchange]
|
routingKey:[[RMQShortstr alloc] init:routingKey]
|
options:options];
|
RMQContentBody *contentBody = [[RMQContentBody alloc] initWithData:body];
|
|
NSData *bodyData = contentBody.amqEncoded;
|
|
NSArray *mergedProperties = [RMQBasicProperties mergeProperties:properties
|
withDefaults:RMQBasicProperties.defaultProperties];
|
RMQContentHeader *contentHeader = [[RMQContentHeader alloc] initWithClassID:publish.classID
|
bodySize:@(bodyData.length)
|
properties:mergedProperties];
|
|
NSArray *contentBodies = [self contentBodiesFromData:bodyData
|
inChunksOf:self.contentBodySize.integerValue];
|
RMQFrameset *frameset = [[RMQFrameset alloc] initWithChannelNumber:self.channelNumber
|
method:publish
|
contentHeader:contentHeader
|
contentBodies:contentBodies];
|
[self.dispatcher sendAsyncFrameset:frameset];
|
return [self.confirmations addPublication];
|
}
|
|
- (void)basicGet:(NSString *)queue
|
options:(RMQBasicGetOptions)options
|
completionHandler:(RMQConsumerDeliveryHandler)userCompletionHandler {
|
[self.dispatcher sendSyncMethod:[[RMQBasicGet alloc] initWithReserved1:[[RMQShort alloc] init:0]
|
queue:[[RMQShortstr alloc] init:queue]
|
options:options]
|
completionHandler:^(RMQFrameset *frameset) {
|
RMQBasicGetOk *getOk = (RMQBasicGetOk *)frameset.method;
|
RMQMessage *message = [[RMQMessage alloc] initWithBody:frameset.contentData
|
consumerTag:@""
|
deliveryTag:@(getOk.deliveryTag.integerValue)
|
redelivered:getOk.options & RMQBasicGetOkRedelivered
|
exchangeName:getOk.exchange.stringValue
|
routingKey:getOk.routingKey.stringValue
|
properties:frameset.contentHeader.properties];
|
userCompletionHandler(message);
|
}];
|
}
|
|
- (void)basicQos:(NSNumber *)count
|
global:(BOOL)isGlobal {
|
[self.dispatcher sendSyncMethod:[[RMQBasicQos alloc] initWithPrefetchCount:count
|
global:isGlobal]
|
completionHandler:^(RMQFrameset *frameset) {
|
if (isGlobal) {
|
self.prefetchCountPerChannel = count;
|
} else {
|
self.prefetchCountPerConsumer = count;
|
}
|
}];
|
}
|
|
- (void)ack:(NSNumber *)deliveryTag
|
options:(RMQBasicAckOptions)options {
|
[self.dispatcher sendAsyncMethod:[[RMQBasicAck alloc] initWithDeliveryTag:[[RMQLonglong alloc] init:deliveryTag.integerValue]
|
options:options]];
|
}
|
|
- (void)ack:(NSNumber *)deliveryTag {
|
[self ack:deliveryTag options:RMQBasicAckNoOptions];
|
}
|
|
- (void)reject:(NSNumber *)deliveryTag
|
options:(RMQBasicRejectOptions)options {
|
[self.dispatcher sendAsyncMethod:[[RMQBasicReject alloc] initWithDeliveryTag:[[RMQLonglong alloc] init:deliveryTag.integerValue]
|
options:options]];
|
}
|
|
- (void)reject:(NSNumber *)deliveryTag {
|
[self reject:deliveryTag options:RMQBasicRejectNoOptions];
|
}
|
|
- (void)nack:(NSNumber *)deliveryTag
|
options:(RMQBasicNackOptions)options {
|
[self.dispatcher sendAsyncMethod:[[RMQBasicNack alloc] initWithDeliveryTag:[[RMQLonglong alloc] init:deliveryTag.integerValue]
|
options:options]];
|
}
|
|
- (void)nack:(NSNumber *)deliveryTag {
|
[self nack:deliveryTag options:RMQBasicNackNoOptions];
|
}
|
|
- (void)exchangeDeclare:(NSString *)name
|
type:(NSString *)type
|
options:(RMQExchangeDeclareOptions)options {
|
[self.dispatcher sendSyncMethod:[[RMQExchangeDeclare alloc] initWithExchange:name
|
type:type
|
options:options]];
|
}
|
|
- (void)exchangeBind:(NSString *)sourceName
|
destination:(NSString *)destinationName
|
routingKey:(NSString *)routingKey {
|
[self.dispatcher sendSyncMethod:[[RMQExchangeBind alloc] initWithDestination:destinationName
|
source:sourceName
|
routingKey:routingKey]];
|
[self.exchangeBindings[sourceName] addObject:@{@"destination": destinationName,
|
@"routing-key": routingKey}];
|
}
|
|
- (void)exchangeUnbind:(NSString *)sourceName
|
destination:(NSString *)destinationName
|
routingKey:(NSString *)routingKey {
|
[self.dispatcher sendSyncMethod:[[RMQExchangeUnbind alloc] initWithDestination:destinationName
|
source:sourceName
|
routingKey:routingKey]];
|
[self.exchangeBindings[sourceName] removeObject:@{@"destination": destinationName,
|
@"routing-key": routingKey}];
|
}
|
|
- (RMQExchange *)fanout:(NSString *)name options:(RMQExchangeDeclareOptions)options {
|
return [self memoizedExchangeDeclare:name type:@"fanout" options:options];
|
}
|
|
- (RMQExchange *)fanout:(NSString *)name {
|
return [self fanout:name options:RMQExchangeDeclareNoOptions];
|
}
|
|
- (RMQExchange *)direct:(NSString *)name options:(RMQExchangeDeclareOptions)options {
|
return [self memoizedExchangeDeclare:name type:@"direct" options:options];
|
}
|
|
- (RMQExchange *)direct:(NSString *)name {
|
return [self direct:name options:RMQExchangeDeclareNoOptions];
|
}
|
|
- (RMQExchange *)topic:(NSString *)name options:(RMQExchangeDeclareOptions)options {
|
return [self memoizedExchangeDeclare:name type:@"topic" options:options];
|
}
|
|
- (RMQExchange *)topic:(NSString *)name {
|
return [self topic:name options:RMQExchangeDeclareNoOptions];
|
}
|
|
- (RMQExchange *)headers:(NSString *)name options:(RMQExchangeDeclareOptions)options {
|
return [self memoizedExchangeDeclare:name type:@"headers" options:options];
|
}
|
|
- (RMQExchange *)headers:(NSString *)name {
|
return [self headers:name options:RMQExchangeDeclareNoOptions];
|
}
|
|
- (void)exchangeDelete:(NSString *)name
|
options:(RMQExchangeDeleteOptions)options {
|
[self.exchanges removeObjectForKey:name];
|
[self.dispatcher sendSyncMethod:[[RMQExchangeDelete alloc] initWithReserved1:[[RMQShort alloc] init:0]
|
exchange:[[RMQShortstr alloc] init:name]
|
options:options]];
|
}
|
|
# pragma mark - RMQFrameHandler
|
|
- (void)handleFrameset:(RMQFrameset *)frameset {
|
if ([frameset.method isKindOfClass:[RMQBasicDeliver class]]) {
|
[self.dispatcher enqueue:^{
|
[self handleBasicDeliver:frameset];
|
}];
|
} else if ([frameset.method isKindOfClass:[RMQBasicCancel class]]) {
|
[self.dispatcher enqueue:^{
|
[self handleBasicCancel:frameset];
|
}];
|
} else if ([frameset.method isKindOfClass:[RMQBasicAck class]]) {
|
[self.dispatcher enqueue:^{
|
[self.confirmations ack:(RMQBasicAck *)frameset.method];
|
}];
|
} else if ([frameset.method isKindOfClass:[RMQBasicNack class]]) {
|
[self.dispatcher enqueue:^{
|
[self.confirmations nack:(RMQBasicNack *)frameset.method];
|
}];
|
} else {
|
[self.dispatcher handleFrameset:frameset];
|
}
|
}
|
|
# pragma mark - Private
|
|
- (void)handleBasicDeliver:(RMQFrameset *)frameset {
|
RMQBasicDeliver *deliver = (RMQBasicDeliver *)frameset.method;
|
RMQConsumer *consumer = self.consumers[deliver.consumerTag.stringValue];
|
if (consumer) {
|
[consumer consume: [[RMQMessage alloc] initWithBody:frameset.contentData
|
consumerTag:deliver.consumerTag.stringValue
|
deliveryTag:@(deliver.deliveryTag.integerValue)
|
redelivered:deliver.options & RMQBasicDeliverRedelivered
|
exchangeName:deliver.exchange.stringValue
|
routingKey:deliver.routingKey.stringValue
|
properties:frameset.contentHeader.properties]];
|
}
|
}
|
|
- (void)handleBasicCancel:(RMQFrameset *)frameset {
|
RMQBasicCancel *cancel = (RMQBasicCancel *)frameset.method;
|
NSString *consumerTag = cancel.consumerTag.stringValue;
|
RMQConsumer *consumer = self.consumers[consumerTag];
|
[consumer handleCancellation];
|
[self.consumers removeObjectForKey:consumerTag];
|
}
|
|
- (RMQExchange *)memoizedExchangeDeclare:(NSString *)name
|
type:(NSString *)type
|
options:(RMQExchangeDeclareOptions)options {
|
RMQExchange *exchange;
|
exchange = self.exchanges[name];
|
if (!exchange) {
|
[self exchangeDeclare:name type:type options:options];
|
exchange = [[RMQExchange alloc] initWithName:name
|
type:type
|
options:options
|
channel:self];
|
self.exchanges[name] = exchange;
|
self.exchangeBindings[name] = [NSMutableSet new];
|
}
|
return exchange;
|
}
|
|
- (RMQQueue *)memoizedQueueDeclare:(NSString *)originalQueueName
|
options:(RMQQueueDeclareOptions)options
|
arguments:(RMQTable *)arguments {
|
NSString *declaredQueueName = [originalQueueName isEqualToString:@""]
|
? [self.nameGenerator generateWithPrefix:@"rmq-objc-client.gen-"]
|
: originalQueueName;
|
|
if (self.queues[declaredQueueName]) {
|
NSError *error = [NSError errorWithDomain:RMQErrorDomain
|
code:RMQErrorChannelQueueNameCollision
|
userInfo:@{NSLocalizedDescriptionKey: @"Name collision when generating unique name."}];
|
[self.delegate channel:self error:error];
|
return nil;
|
} else {
|
RMQQueue *q = [[RMQQueue alloc] initWithName:declaredQueueName
|
options:options
|
arguments:arguments
|
channel:(id<RMQChannel>)self];
|
|
RMQQueueDeclare *method = [[RMQQueueDeclare alloc] initWithQueue:declaredQueueName
|
options:options
|
arguments:arguments];
|
[self.dispatcher sendSyncMethod:method];
|
|
self.queues[q.name] = q;
|
self.queueBindings[q.name] = [NSMutableSet new];
|
return q;
|
}
|
}
|
|
- (NSArray *)contentBodiesFromData:(NSData *)data inChunksOf:(NSUInteger)chunkSize {
|
NSMutableArray *bodies = [NSMutableArray new];
|
NSUInteger chunkCount = data.length / chunkSize;
|
for (int i = 0; i < chunkCount; i++) {
|
NSUInteger offset = i * chunkSize;
|
NSData *subData = [data subdataWithRange:NSMakeRange(offset, chunkSize)];
|
RMQContentBody *body = [[RMQContentBody alloc] initWithData:subData];
|
[bodies addObject:body];
|
}
|
NSUInteger lastChunkSize = data.length % chunkSize;
|
if (lastChunkSize > 0) {
|
NSData *lastData = [data subdataWithRange:NSMakeRange(data.length - lastChunkSize, lastChunkSize)];
|
[bodies addObject:[[RMQContentBody alloc] initWithData:lastData]];
|
}
|
return bodies;
|
}
|
|
- (void)recoverPrefetch {
|
if (self.prefetchCountPerConsumer) {
|
[self basicQos:self.prefetchCountPerConsumer global:NO];
|
}
|
if (self.prefetchCountPerChannel) {
|
[self basicQos:self.prefetchCountPerChannel global:YES];
|
}
|
}
|
|
- (void)recoverConfirmations {
|
if (self.confirmations.isEnabled) {
|
[self.confirmations recover];
|
[self.dispatcher sendSyncMethod:[RMQConfirmSelect new]];
|
}
|
}
|
|
- (void)recoverExchanges {
|
for (RMQExchange *exchange in self.exchanges.allValues) {
|
[self exchangeDeclare:exchange.name type:exchange.type options:exchange.options];
|
}
|
}
|
|
- (void)recoverExchangeBindings {
|
for (RMQExchange *exchange in self.exchanges.allValues) {
|
for (NSDictionary *binding in [self.exchangeBindings[exchange.name] copy]) {
|
[self exchangeBind:exchange.name destination:binding[@"destination"] routingKey:binding[@"routing-key"]];
|
}
|
}
|
}
|
|
- (void)recoverQueuesAndTheirBindings {
|
for (RMQQueue *queue in self.queues.allValues) {
|
[self.dispatcher sendSyncMethod:[[RMQQueueDeclare alloc] initWithQueue:queue.name
|
options:queue.options
|
arguments:queue.arguments]];
|
for (NSDictionary *binding in [self.queueBindings[queue.name] copy]) {
|
[self queueBind:queue.name exchange:binding[@"exchange"] routingKey:binding[@"routing-key"]];
|
}
|
}
|
}
|
|
- (void)recoverConsumers {
|
for (RMQConsumer *consumer in self.consumers.allValues) {
|
[self.dispatcher sendSyncMethod:[[RMQBasicConsume alloc] initWithQueue:consumer.queueName
|
consumerTag:consumer.tag
|
options:consumer.options]];
|
}
|
}
|
|
@end
|
