/*
 * 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 "FIRMessagingConnection.h"

#import "Protos/GtalkCore.pbobjc.h"
#import "Protos/GtalkExtensions.pbobjc.h"

#import "FIRMessaging.h"
#import "FIRMessagingDataMessageManager.h"
#import "FIRMessagingDefines.h"
#import "FIRMessagingLogger.h"
#import "FIRMessagingRmqManager.h"
#import "FIRMessagingSecureSocket.h"
#import "FIRMessagingUtilities.h"
#import "FIRMessagingVersionUtilities.h"
#import "FIRMessaging_Private.h"

static NSInteger const kIqSelectiveAck = 12;
static NSInteger const kIqStreamAck = 13;
static int const kInvalidStreamId = -1;
// Threshold for number of messages removed that we will ack, for short lived connections
static int const kMessageRemoveAckThresholdCount = 5;

static NSTimeInterval const kHeartbeatInterval = 30.0;
static NSTimeInterval const kConnectionTimeout = 20.0;
static int32_t const kAckingInterval = 10;

static NSString *const kUnackedS2dIdKey = @"FIRMessagingUnackedS2dIdKey";
static NSString *const kAckedS2dIdMapKey = @"FIRMessagingAckedS2dIdMapKey";

static NSString *const kRemoteFromAddress = @"from";

@interface FIRMessagingD2SInfo : NSObject

@property(nonatomic, readwrite, assign) int streamId;
@property(nonatomic, readwrite, strong) NSString *d2sID;
- (instancetype)initWithStreamId:(int)streamId d2sId:(NSString *)d2sID;

@end

@implementation FIRMessagingD2SInfo

- (instancetype)initWithStreamId:(int)streamId d2sId:(NSString *)d2sID {
  self = [super init];
  if (self) {
    _streamId = streamId;
    _d2sID = [d2sID copy];
  }
  return self;
}

- (BOOL)isEqual:(id)object {
  if ([object isKindOfClass:[self class]]) {
    FIRMessagingD2SInfo *other = (FIRMessagingD2SInfo *)object;
    return self.streamId == other.streamId && [self.d2sID isEqualToString:other.d2sID];
  }
  return NO;
}

- (NSUInteger)hash {
  return [self.d2sID hash];
}

@end

@interface FIRMessagingConnection ()<FIRMessagingSecureSocketDelegate>

@property(nonatomic, readwrite, weak) FIRMessagingRmqManager *rmq2Manager;
@property(nonatomic, readwrite, weak) FIRMessagingDataMessageManager *dataMessageManager;

@property(nonatomic, readwrite, assign) FIRMessagingConnectionState state;
@property(nonatomic, readwrite, copy) NSString *host;
@property(nonatomic, readwrite, assign) NSUInteger port;

@property(nonatomic, readwrite, strong) NSString *authId;
@property(nonatomic, readwrite, strong) NSString *token;

@property(nonatomic, readwrite, strong) FIRMessagingSecureSocket *socket;

@property(nonatomic, readwrite, assign) int64_t lastLoginServerTimestamp;
@property(nonatomic, readwrite, assign) int lastStreamIdAcked;
@property(nonatomic, readwrite, assign) int inStreamId;
@property(nonatomic, readwrite, assign) int outStreamId;

@property(nonatomic, readwrite, strong) NSMutableArray *unackedS2dIds;
@property(nonatomic, readwrite, strong) NSMutableDictionary *ackedS2dMap;
@property(nonatomic, readwrite, strong) NSMutableArray *d2sInfos;
// ttl=0 messages that need to be sent as soon as we establish a connection
@property(nonatomic, readwrite, strong) NSMutableArray *sendOnConnectMessages;

@property(nonatomic, readwrite, strong) NSRunLoop *runLoop;

@end


@implementation FIRMessagingConnection;

- (instancetype)initWithAuthID:(NSString *)authId
                         token:(NSString *)token
                          host:(NSString *)host
                          port:(NSUInteger)port
                       runLoop:(NSRunLoop *)runLoop
                   rmq2Manager:(FIRMessagingRmqManager *)rmq2Manager
                    fcmManager:(FIRMessagingDataMessageManager *)dataMessageManager {
  self = [super init];
  if (self) {
    _authId = [authId copy];
    _token = [token copy];
    _host = [host copy];
    _port = port;
    _runLoop = runLoop;
    _rmq2Manager = rmq2Manager;
    _dataMessageManager = dataMessageManager;

    _d2sInfos = [NSMutableArray array];

    _unackedS2dIds = [NSMutableArray arrayWithArray:[_rmq2Manager unackedS2dRmqIds]];
    _ackedS2dMap = [NSMutableDictionary dictionary];
    _sendOnConnectMessages = [NSMutableArray array];
  }
  return self;
}

- (NSString *)description {
  return [NSString stringWithFormat:@"host: %@, port: %lu, stream id in: %d, stream id out: %d",
          self.host,
          _FIRMessaging_UL(self.port),
          self.inStreamId,
          self.outStreamId];
}

- (void)signIn {
  _FIRMessagingDevAssert(self.state == kFIRMessagingConnectionNotConnected, @"Invalid connection state.");
  if (self.state != kFIRMessagingConnectionNotConnected) {
    return;
  }

  // break it up for testing
  [self setupConnectionSocket];
  [self connectToSocket:self.socket];
}

- (void)setupConnectionSocket {
  self.socket = [[FIRMessagingSecureSocket alloc] init];
  self.socket.delegate = self;
}

- (void)connectToSocket:(FIRMessagingSecureSocket *)socket {
  self.state = kFIRMessagingConnectionConnecting;
  FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection000,
                          @"Start connecting to FIRMessaging service.");
  [socket connectToHost:self.host port:self.port onRunLoop:self.runLoop];
}

- (void)signOut {
  // Clear the list of messages to be sent on connect. This will only
  // have messages in it if an error happened before receiving the LoginResponse.
  [self.sendOnConnectMessages removeAllObjects];

  if (self.state == kFIRMessagingConnectionSignedIn) {
    [self sendClose];
  }
  if (self.state != kFIRMessagingConnectionNotConnected) {
    [self disconnect];
  }
}

- (void)teardown {
  if (self.state != kFIRMessagingConnectionNotConnected) {
    [self disconnect];
  }
}

#pragma mark - FIRMessagingSecureSocketDelegate

- (void)secureSocketDidConnect:(FIRMessagingSecureSocket *)socket {
  self.state = kFIRMessagingConnectionConnected;
  self.lastStreamIdAcked = 0;
  self.inStreamId = 0;
  self.outStreamId = 0;

  FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection001,
                          @"Connected to FIRMessaging service.");
  [self resetUnconfirmedAcks];
  [self sendLoginRequest:self.authId token:self.token];
}

- (void)didDisconnectWithSecureSocket:(FIRMessagingSecureSocket *)socket {
  _FIRMessagingDevAssert(self.socket == socket, @"Invalid socket");
  _FIRMessagingDevAssert(self.socket.state == kFIRMessagingSecureSocketClosed, @"Socket already closed");

  FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection002,
                          @"Secure socket disconnected from FIRMessaging service.");
  [self disconnect];
  [self.delegate connection:self didCloseForReason:kFIRMessagingConnectionCloseReasonSocketDisconnected];
}

- (void)secureSocket:(FIRMessagingSecureSocket *)socket
      didReceiveData:(NSData *)data
             withTag:(int8_t)tag {
  if (tag < 0) {
    // Invalid proto tag
    return;
  }

  Class klassForTag = FIRMessagingGetClassForTag((FIRMessagingProtoTag)tag);
  if ([klassForTag isSubclassOfClass:[NSNull class]]) {
    FIRMessagingLoggerError(kFIRMessagingMessageCodeConnection003, @"Invalid tag %d for proto",
                            tag);
    return;
  }

  GPBMessage *proto = [klassForTag parseFromData:data error:NULL];
  if (tag == kFIRMessagingProtoTagLoginResponse && self.state != kFIRMessagingConnectionConnected) {
    FIRMessagingLoggerDebug(
        kFIRMessagingMessageCodeConnection004,
        @"Should not receive generated message when the connection is not connected.");
    return;
  } else if (tag != kFIRMessagingProtoTagLoginResponse && self.state != kFIRMessagingConnectionSignedIn) {
    FIRMessagingLoggerDebug(
        kFIRMessagingMessageCodeConnection005,
        @"Should not receive generated message when the connection is not signed in.");
    return;
  }

  // If traffic is received after a heartbeat it is safe to assume the connection is healthy.
  [self cancelConnectionTimeoutTask];
  [self performSelector:@selector(sendHeartbeatPing)
             withObject:nil
             afterDelay:kHeartbeatInterval];

  [self willProcessProto:proto];
  switch (tag) {
    case kFIRMessagingProtoTagLoginResponse:
      [self didReceiveLoginResponse:(GtalkLoginResponse *)proto];
      break;
    case kFIRMessagingProtoTagDataMessageStanza:
      [self didReceiveDataMessageStanza:(GtalkDataMessageStanza *)proto];
      break;
    case kFIRMessagingProtoTagHeartbeatPing:
      [self didReceiveHeartbeatPing:(GtalkHeartbeatPing *)proto];
      break;
    case kFIRMessagingProtoTagHeartbeatAck:
      [self didReceiveHeartbeatAck:(GtalkHeartbeatAck *)proto];
      break;
    case kFIRMessagingProtoTagClose:
      [self didReceiveClose:(GtalkClose *)proto];
      break;
    case kFIRMessagingProtoTagIqStanza:
      [self handleIqStanza:(GtalkIqStanza *)proto];
      break;
    default:
      [self didReceiveUnhandledProto:proto];
      break;
  }
}

// Called from secure socket once we have send the proto with given rmqId over the wire
// since we are mostly concerned with user facing messages which certainly have a rmqId
// we can retrieve them from the Rmq if necessary to look at stuff but for now we just
// log it.
- (void)secureSocket:(FIRMessagingSecureSocket *)socket
 didSendProtoWithTag:(int8_t)tag
               rmqId:(NSString *)rmqId {
  // log the message
  [self logMessage:rmqId messageType:tag isOut:YES];
}

#pragma mark - FIRMessagingTestConnection

- (void)sendProto:(GPBMessage *)proto {
  FIRMessagingProtoTag tag = FIRMessagingGetTagForProto(proto);
  if (tag == kFIRMessagingProtoTagLoginRequest && self.state != kFIRMessagingConnectionConnected) {
    FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection006,
                            @"Cannot send generated message when the connection is not connected.");
    return;
  } else if (tag != kFIRMessagingProtoTagLoginRequest && self.state != kFIRMessagingConnectionSignedIn) {
    FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection007,
                            @"Cannot send generated message when the connection is not signed in.");
    return;
  }

  _FIRMessagingDevAssert(self.socket != nil, @"Socket shouldn't be nil");
  if (self.socket == nil) {
    return;
  }

  [self willSendProto:proto];

  [self.socket sendData:proto.data withTag:tag rmqId:FIRMessagingGetRmq2Id(proto)];
}

- (void)sendOnConnectOrDrop:(GPBMessage *)message {
  if (self.state == kFIRMessagingConnectionSignedIn) {
    // If a connection has already been established, send normally
    [self sendProto:message];
  } else {
    // Otherwise add them to the list of messages to send after login
    [self.sendOnConnectMessages addObject:message];
  }
}

+ (GtalkLoginRequest *)loginRequestWithToken:(NSString *)token authID:(NSString *)authID {
  GtalkLoginRequest *login = [[GtalkLoginRequest alloc] init];
  login.accountId = 1000000;
  login.authService = GtalkLoginRequest_AuthService_AndroidId;
  login.authToken = token;
  login.id_p = [NSString stringWithFormat:@"%@-%@", @"ios", FIRMessagingCurrentLibraryVersion()];
  login.domain = @"mcs.android.com";
  login.deviceId = [NSString stringWithFormat:@"android-%llx", authID.longLongValue];
  login.networkType = [self currentNetworkType];
  login.resource = authID;
  login.user = authID;
  login.useRmq2 = YES;
  login.lastRmqId = 1; // Sending not enabled yet so this stays as 1.
  return login;
}

+ (int32_t)currentNetworkType {
  // http://developer.android.com/reference/android/net/ConnectivityManager.html
  int32_t fcmNetworkType;
  FIRMessagingNetworkStatus type = [[FIRMessaging messaging] networkType];
  switch (type) {
    case kFIRMessagingReachabilityReachableViaWiFi:
      fcmNetworkType = 1;
      break;

    case kFIRMessagingReachabilityReachableViaWWAN:
      fcmNetworkType = 0;
      break;

    default:
      fcmNetworkType = -1;
      break;
  }
  return fcmNetworkType;
}

- (void)sendLoginRequest:(NSString *)authId
                   token:(NSString *)token {
  GtalkLoginRequest *login = [[self class] loginRequestWithToken:token authID:authId];

  // clear the messages sent during last connection
  if ([self.d2sInfos count]) {
    [self.d2sInfos removeAllObjects];
  }

  if (self.unackedS2dIds.count > 0) {
    FIRMessagingLoggerDebug(
        kFIRMessagingMessageCodeConnection008,
        @"There are unacked persistent Ids in the login request: %@",
        [self.unackedS2dIds.description stringByReplacingOccurrencesOfString:@"%"
                                                                  withString:@"%%"]);
  }
  // Send out acks.
  for (NSString *unackedPersistentS2dId in self.unackedS2dIds) {
    [login.receivedPersistentIdArray addObject:unackedPersistentS2dId];
  }

  GtalkSetting *setting = [[GtalkSetting alloc] init];
  setting.name = @"new_vc";
  setting.value = @"1";
  [login.settingArray addObject:setting];

  [self sendProto:login];
}

- (void)sendHeartbeatAck {
  [self sendProto:[[GtalkHeartbeatAck alloc] init]];
}

- (void)sendHeartbeatPing {
  // cancel the previous heartbeat request.
  [NSObject cancelPreviousPerformRequestsWithTarget:self
                                           selector:@selector(sendHeartbeatPing)
                                             object:nil];
  [self scheduleConnectionTimeoutTask];
  [self sendProto:[[GtalkHeartbeatPing alloc] init]];
}

+ (GtalkIqStanza *)createStreamAck {
  GtalkIqStanza *iq = [[GtalkIqStanza alloc] init];
  iq.type = GtalkIqStanza_IqType_Set;
  iq.id_p = @"";
  GtalkExtension *ext = [[GtalkExtension alloc] init];
  ext.id_p = kIqStreamAck;
  ext.data_p = @"";
  iq.extension = ext;
  return iq;
}

- (void)sendStreamAck {
  GtalkIqStanza *iq = [[self class] createStreamAck];
  [self sendProto:iq];
}

- (void)sendClose {
  [self sendProto:[[GtalkClose alloc] init]];
}

- (void)handleIqStanza:(GtalkIqStanza *)iq {
  if (iq.hasExtension) {
    if (iq.extension.id_p == kIqStreamAck) {
      [self didReceiveStreamAck:iq];
      return;
    }
    if (iq.extension.id_p == kIqSelectiveAck) {
      [self didReceiveSelectiveAck:iq];
      return;
    }
    FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection009, @"Unknown ack extension id %d.",
                            iq.extension.id_p);
  } else {
    FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection010, @"Ip stanza without extension.");
  }
  [self didReceiveUnhandledProto:iq];
}

- (void)didReceiveLoginResponse:(GtalkLoginResponse *)loginResponse {
  if (loginResponse.hasError) {
    FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection011,
                            @"Login error with type: %@, message: %@.", loginResponse.error.type,
                            loginResponse.error.message);
    return;
  }
  FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection012, @"Logged onto MCS service.");
  // We sent the persisted list of unack'd messages with login so we can assume they have been ack'd
  // by the server.
  _FIRMessagingDevAssert(self.unackedS2dIds.count == 0, @"No ids present");
  _FIRMessagingDevAssert(self.outStreamId == 1, @"Login should be the first stream id");

  self.state = kFIRMessagingConnectionSignedIn;
  self.lastLoginServerTimestamp = loginResponse.serverTimestamp;
  [self.delegate didLoginWithConnection:self];
  [self sendHeartbeatPing];

  // Add all the TTL=0 messages on connect
  for (GPBMessage *message in self.sendOnConnectMessages) {
    [self sendProto:message];
  }
  [self.sendOnConnectMessages removeAllObjects];
}

- (void)didReceiveHeartbeatPing:(GtalkHeartbeatPing *)heartbeatPing {
  [self sendHeartbeatAck];
}

- (void)didReceiveHeartbeatAck:(GtalkHeartbeatAck *)heartbeatAck {
}

- (void)didReceiveDataMessageStanza:(GtalkDataMessageStanza *)dataMessageStanza {
  // TODO: Maybe add support raw data later
  [self.delegate connectionDidRecieveMessage:dataMessageStanza];
}

- (void)didReceiveUnhandledProto:(GPBMessage *)proto {
  FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection013, @"Received unhandled proto");
}

- (void)didReceiveStreamAck:(GtalkIqStanza *)iq {
  // Server received some stuff from us we don't really need to do anything special
}

- (void)didReceiveSelectiveAck:(GtalkIqStanza *)iq {
  GtalkExtension *extension = iq.extension;
  if (extension) {
    int extensionId = extension.id_p;
    if (extensionId == kIqSelectiveAck) {

      NSString *dataString = extension.data_p;
      GtalkSelectiveAck *selectiveAck = [[GtalkSelectiveAck alloc] init];
      [selectiveAck mergeFromData:[dataString dataUsingEncoding:NSUTF8StringEncoding]
                extensionRegistry:nil];

      NSArray <NSString *>*acks = [selectiveAck idArray];

      // we've received ACK's
      [self.delegate connectionDidReceiveAckForRmqIds:acks];

      // resend unacked messages
      [self.dataMessageManager resendMessagesWithConnection:self];
    }
  }
}

- (void)didReceiveClose:(GtalkClose *)close {
  [self disconnect];
}

- (void)willProcessProto:(GPBMessage *)proto {
  self.inStreamId++;

  if ([proto isKindOfClass:GtalkDataMessageStanza.class]) {
    FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection014,
                            @"RMQ: Receiving %@ with rmq_id: %@ incoming stream Id: %d",
                            proto.class, FIRMessagingGetRmq2Id(proto), self.inStreamId);
  } else {
    FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection015,
                            @"RMQ: Receiving %@ with incoming stream Id: %d.", proto.class,
                            self.inStreamId);
  }
  int streamId = FIRMessagingGetLastStreamId(proto);
  if (streamId != kInvalidStreamId) {
    // confirm the D2S messages that were sent by us
    [self confirmAckedD2sIdsWithStreamId:streamId];

    // We can now confirm that our ack was received by the server and start our unack'd list fresh
    // with the proto we just received.
    [self confirmAckedS2dIdsWithStreamId:streamId];
  }
  NSString *rmq2Id = FIRMessagingGetRmq2Id(proto);
  if (rmq2Id != nil) {
    FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection016,
                            @"RMQ: Add unacked persistent Id: %@.",
                            [rmq2Id stringByReplacingOccurrencesOfString:@"%" withString:@"%%"]);
    [self.unackedS2dIds addObject:rmq2Id];
    [self.rmq2Manager saveS2dMessageWithRmqId:rmq2Id]; // RMQ save
  }
  BOOL explicitAck = ([proto isKindOfClass:[GtalkDataMessageStanza class]] &&
                      [(GtalkDataMessageStanza *)proto immediateAck]);
  // If we have not sent anything and the ack threshold has been reached then explicitly send one
  // to notify the server that we have received messages.
  if (self.inStreamId - self.lastStreamIdAcked >= kAckingInterval || explicitAck) {
    [self sendStreamAck];
  }
}

- (void)willSendProto:(GPBMessage *)proto {
  self.outStreamId++;

  NSString *rmq2Id = FIRMessagingGetRmq2Id(proto);
  if ([rmq2Id length]) {
    FIRMessagingD2SInfo *d2sInfo = [[FIRMessagingD2SInfo alloc] initWithStreamId:self.outStreamId d2sId:rmq2Id];
    [self.d2sInfos addObject:d2sInfo];
  }

  // each time we send a d2s message, it acks previously received
  // s2d messages via the last (s2d) stream id received.

  FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection017,
                          @"RMQ: Sending %@ with outgoing stream Id: %d.", proto.class,
                          self.outStreamId);
  // We have received messages since last time we sent something - send ack info to server.
  if (self.inStreamId > self.lastStreamIdAcked) {
    FIRMessagingSetLastStreamId(proto, self.inStreamId);
    self.lastStreamIdAcked = self.inStreamId;
  }

  if (self.unackedS2dIds.count > 0) {
    // Move all 'unack'd' messages to the ack'd map so they can be removed once the
    // ack is confirmed.
    NSArray *ackedS2dIds = [NSArray arrayWithArray:self.unackedS2dIds];
    FIRMessagingLoggerDebug(
        kFIRMessagingMessageCodeConnection018, @"RMQ: Mark persistent Ids as acked: %@.",
        [ackedS2dIds.description stringByReplacingOccurrencesOfString:@"%" withString:@"%%"]);
    [self.unackedS2dIds removeAllObjects];
    self.ackedS2dMap[[@(self.outStreamId) stringValue]] = ackedS2dIds;
  }
}

#pragma mark - Private

/**
 * This processes the s2d message received in reference to the d2s messages
 * that we have sent before.
 */
- (void)confirmAckedD2sIdsWithStreamId:(int)lastReceivedStreamId {
  NSMutableArray *d2sIdsAcked = [NSMutableArray array];
  for (FIRMessagingD2SInfo *d2sInfo in self.d2sInfos) {
    if (lastReceivedStreamId < d2sInfo.streamId) {
      break;
    }
    [d2sIdsAcked addObject:d2sInfo];
  }

  NSMutableArray *rmqIds = [NSMutableArray arrayWithCapacity:[d2sIdsAcked count]];
  // remove ACK'ed messages
  for (FIRMessagingD2SInfo *d2sInfo in d2sIdsAcked) {
    if ([d2sInfo.d2sID length]) {
      [rmqIds addObject:d2sInfo.d2sID];
    }
    [self.d2sInfos removeObject:d2sInfo];
  }
  [self.delegate connectionDidReceiveAckForRmqIds:rmqIds];
  int count = [self.delegate connectionDidReceiveAckForRmqIds:rmqIds];
  if (kMessageRemoveAckThresholdCount > 0 && count >= kMessageRemoveAckThresholdCount) {
    // For short lived connections, if a large number of messages are removed, send an
    // ack straight away so the server knows that this message was received.
    [self sendStreamAck];
  }
}

/**
 * Called when a stream ACK or a selective ACK are received - this indicates the message has
 * been received by MCS.
 */
- (void)didReceiveAckForRmqIds:(NSArray *)rmqIds {
  // TODO: let the user know that the following messages were received by the server
}

- (void)confirmAckedS2dIdsWithStreamId:(int)lastReceivedStreamId {
  // If the server hasn't received the streamId yet.
  FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection019,
                          @"RMQ: Server last received stream Id: %d.", lastReceivedStreamId);
  if (lastReceivedStreamId < self.outStreamId) {
    // TODO: This could be a good indicator that we need to re-send something (acks)?
    FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection020,
                            @"RMQ: There are unsent messages that should be send...\n"
                             "server received: %d\nlast stream id sent: %d",
                            lastReceivedStreamId, self.outStreamId);
  }

  NSSet *ackedStreamIds =
    [self.ackedS2dMap keysOfEntriesPassingTest:^BOOL(id key, id obj, BOOL *stop) {
      NSString *streamId = key;
      return streamId.intValue <= lastReceivedStreamId;
    }];
  NSMutableArray *s2dIdsToDelete = [NSMutableArray array];

  for (NSString *streamId in ackedStreamIds) {
    NSArray *ackedS2dIds = self.ackedS2dMap[streamId];
    if (ackedS2dIds.count > 0) {
      FIRMessagingLoggerDebug(
          kFIRMessagingMessageCodeConnection021,
          @"RMQ: Mark persistent Ids as confirmed by stream id %@: %@.", streamId,
          [ackedS2dIds.description stringByReplacingOccurrencesOfString:@"%" withString:@"%%"]);
      [self.ackedS2dMap removeObjectForKey:streamId];
    }

    [s2dIdsToDelete addObjectsFromArray:ackedS2dIds];
  }

  // clean up s2d ids that the server knows we've received.
  // we let the server know via a s2d last stream id received in a
  // d2s message. the server lets us know it has received our d2s
  // message via a d2s last stream id received in a s2d message.
  [self.rmq2Manager removeS2dIds:s2dIdsToDelete];
}

- (void)resetUnconfirmedAcks {
  [self.ackedS2dMap enumerateKeysAndObjectsUsingBlock:^(id key, id obj, BOOL *stop) {
    [self.unackedS2dIds addObjectsFromArray:obj];
  }];
  [self.ackedS2dMap removeAllObjects];
}

- (void)disconnect {
  _FIRMessagingDevAssert(self.state != kFIRMessagingConnectionNotConnected, @"Connection already not connected");
  // cancel pending timeout tasks.
  [self cancelConnectionTimeoutTask];
  // cancel pending heartbeat.
  [NSObject cancelPreviousPerformRequestsWithTarget:self
                                           selector:@selector(sendHeartbeatPing)
                                             object:nil];
  // Unset the delegate. FIRMessagingConnection will not receive further events from the socket from now on.
  self.socket.delegate = nil;
  [self.socket disconnect];
  self.state = kFIRMessagingConnectionNotConnected;
}

- (void)connectionTimedOut {
  FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection022,
                          @"Connection to FIRMessaging service timed out.");
  [self disconnect];
  [self.delegate connection:self didCloseForReason:kFIRMessagingConnectionCloseReasonTimeout];
}

- (void)scheduleConnectionTimeoutTask {
  // cancel the previous heartbeat timeout event and schedule a new one.
  [self cancelConnectionTimeoutTask];
  [self performSelector:@selector(connectionTimedOut)
             withObject:nil
             afterDelay:[self connectionTimeoutInterval]];
}

- (void)cancelConnectionTimeoutTask {
  // cancel pending timeout tasks.
  [NSObject cancelPreviousPerformRequestsWithTarget:self
                                           selector:@selector(connectionTimedOut)
                                             object:nil];
}

- (void)logMessage:(NSString *)description messageType:(int)messageType isOut:(BOOL)isOut {
  messageType = isOut ? -messageType : messageType;
  FIRMessagingLoggerDebug(kFIRMessagingMessageCodeConnection023,
                          @"Send msg: %@ type: %d inStreamId: %d outStreamId: %d", description,
                          messageType, self.inStreamId, self.outStreamId);
}

- (NSTimeInterval)connectionTimeoutInterval {
  return kConnectionTimeout;
}

@end