Delay re-enablement of channel dispatcher

This gives some leeway for the channel to recover before commands start
flowing again.

There ought to be a way around using a leeway, but my experiments with
queuing properly aren't up to scratch yet.

[#119713047]

Author: camelpunch

RMQClient/RMQMultipleChannelAllocator.m

@@ -65,6 +65,7 @@ @interface RMQMultipleChannelAllocator ()
 @property (nonatomic, readwrite) NSMutableDictionary *channels;
 @property (nonatomic, readwrite) NSNumber *syncTimeout;
 @property (nonatomic, readwrite) RMQProcessInfoNameGenerator *nameGenerator;
+@property (nonatomic, readwrite) NSNumber *dispatcherReenableDelay;
 @end
 
 @implementation RMQMultipleChannelAllocator
@@ -78,6 +79,7 @@ - (instancetype)initWithChannelSyncTimeout:(NSNumber *)syncTimeout {
         self.sender = nil;
         self.syncTimeout = syncTimeout;
         self.nameGenerator = [RMQProcessInfoNameGenerator new];
+        self.dispatcherReenableDelay = @1;
     }
     return self;
 }
@@ -148,14 +150,19 @@ - (void)unsafeReleaseChannelNumber:(NSNumber *)channelNumber {
 }
 
 - (RMQAllocatedChannel *)allocatedChannel:(NSUInteger)channelNumber {
-    RMQGCDSerialQueue *commandQueue = [self suspendedDispatchQueue:channelNumber
-                                                              type:@"commands"];
-    RMQGCDSerialQueue *recoveryQueue = [self suspendedDispatchQueue:channelNumber
-                                                               type:@"recovery"];
+    RMQGCDSerialQueue *commandQueue = [self serialQueue:channelNumber type:@"commands"];
+    [commandQueue suspend];
+    RMQGCDSerialQueue *recoveryQueue = [self serialQueue:channelNumber type:@"recovery"];
+    [recoveryQueue suspend];
+    RMQGCDSerialQueue *enablementQueue = [self serialQueue:channelNumber type:@"enablement"];
+
     RMQSuspendResumeDispatcher *dispatcher = [[RMQSuspendResumeDispatcher alloc] initWithSender:self.sender
-                                                                                   commandQueue:commandQueue];
+                                                                                   commandQueue:commandQueue
+                                                                                enablementQueue:enablementQueue
+                                                                                    enableDelay:self.dispatcherReenableDelay];
     RMQSuspendResumeDispatcher *recoveryDispatcher = [[RMQSuspendResumeDispatcher alloc] initWithSender:self.sender
                                                                                            commandQueue:recoveryQueue];
+
     RMQAllocatedChannel *ch = [[RMQAllocatedChannel alloc] init:@(channelNumber)
                                                 contentBodySize:@(self.sender.frameMax.integerValue - RMQEmptyFrameSize)
                                                      dispatcher:dispatcher
@@ -175,11 +182,9 @@ - (BOOL)atMaxIndex {
     return self.channelNumber == RMQChannelLimit;
 }
 
-- (RMQGCDSerialQueue *)suspendedDispatchQueue:(UInt16)channelNumber
-                                         type:(NSString *)type {
-    RMQGCDSerialQueue *serialQueue = [[RMQGCDSerialQueue alloc] initWithName:[NSString stringWithFormat:@"channel %d (%@)", channelNumber, type]];
-    [serialQueue suspend];
-    return serialQueue;
+- (RMQGCDSerialQueue *)serialQueue:(UInt16)channelNumber
+                              type:(NSString *)type {
+    return [[RMQGCDSerialQueue alloc] initWithName:[NSString stringWithFormat:@"channel %d (%@)", channelNumber, type]];
 }
 
 @end

RMQClient/RMQSuspendResumeDispatcher.h

@@ -57,6 +57,11 @@
 
 @interface RMQSuspendResumeDispatcher : NSObject <RMQDispatcher>
 
+- (instancetype)initWithSender:(id<RMQSender>)sender
+                  commandQueue:(id<RMQLocalSerialQueue>)commandQueue
+               enablementQueue:(id<RMQLocalSerialQueue>)enablementQueue
+                   enableDelay:(NSNumber *)enableDelay;
+
 - (instancetype)initWithSender:(id<RMQSender>)sender
                   commandQueue:(id<RMQLocalSerialQueue>)commandQueue;
 

RMQClient/RMQSuspendResumeDispatcher.m

@@ -63,6 +63,8 @@ @interface RMQSuspendResumeDispatcher ()
 @property (nonatomic, readwrite) id<RMQSender> sender;
 @property (nonatomic, readwrite) RMQFramesetValidator *validator;
 @property (nonatomic, readwrite) id<RMQLocalSerialQueue> commandQueue;
+@property (nonatomic, readwrite) id<RMQLocalSerialQueue> enablementQueue;
+@property (nonatomic, readwrite) NSNumber *enableDelay;
 @property (nonatomic, readwrite) id<RMQConnectionDelegate> delegate;
 @property (nonatomic, readwrite) DispatcherState state;
 @property (nonatomic, readwrite) BOOL disabled;
@@ -71,19 +73,28 @@ @interface RMQSuspendResumeDispatcher ()
 @implementation RMQSuspendResumeDispatcher
 
 - (instancetype)initWithSender:(id<RMQSender>)sender
-                  commandQueue:(id<RMQLocalSerialQueue>)commandQueue {
+                  commandQueue:(id<RMQLocalSerialQueue>)commandQueue
+               enablementQueue:(id<RMQLocalSerialQueue>)enablementQueue
+                   enableDelay:(NSNumber *)enableDelay {
     self = [super init];
     if (self) {
         self.channel = nil;
         self.sender = sender;
         self.validator = [RMQFramesetValidator new];
         self.commandQueue = commandQueue;
+        self.enablementQueue = enablementQueue;
+        self.enableDelay = enableDelay;
         self.state = DispatcherStateOpen;
         self.disabled = NO;
     }
     return self;
 }
 
+- (instancetype)initWithSender:(id<RMQSender>)sender
+                  commandQueue:(id<RMQLocalSerialQueue>)commandQueue {
+    return [self initWithSender:sender commandQueue:commandQueue enablementQueue:nil enableDelay:@0];
+}
+
 - (void)activateWithChannel:(id<RMQChannel>)channel
                    delegate:(id<RMQConnectionDelegate>)delegate {
     self.channel = channel;
@@ -175,8 +186,10 @@ - (void)disable {
 }
 
 - (void)enable {
-    self.disabled = NO;
-    [self.commandQueue resume];
+    [self.enablementQueue delayedBy:self.enableDelay enqueue:^{
+        self.disabled = NO;
+        [self.commandQueue resume];
+    }];
 }
 
 - (void)handleFrameset:(RMQFrameset *)frameset {

RMQClientTests/RMQSuspendResumeDispatcherTest.swift

@@ -68,15 +68,18 @@ class RMQSuspendResumeDispatcherTest: XCTestCase {
 
     func testDisableSuspendsCommandQueueAndPreventsFramesetHandlingFromResuming() {
         let q = FakeSerialQueue()
-        let dispatcher = RMQSuspendResumeDispatcher(sender: SenderSpy(), commandQueue: q)
+        let eq = FakeSerialQueue()
+        let dispatcher = RMQSuspendResumeDispatcher(sender: SenderSpy(), commandQueue: q, enablementQueue: eq, enableDelay: 3)
 
         dispatcher.disable()
         XCTAssertTrue(q.suspended)
 
         dispatcher.handleFrameset(RMQFrameset(channelNumber: 1, method: MethodFixtures.basicQosOk()))
-        XCTAssertTrue(q.suspended)
-
         dispatcher.enable()
+
+        XCTAssertTrue(q.suspended)
+        XCTAssertEqual(3, eq.enqueueDelay)
+        try! eq.step()
         XCTAssertFalse(q.suspended)
     }