import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
/** Reads the Ticl state from persistent storage (if any) and calls {@code startInternal}. */
private void scheduleStartAfterReadingStateBlob() {
  storage.readKey(CLIENT_TOKEN_KEY, new Callback<SimplePair<Status, byte[]>>() {
    @Override
    public void accept(final SimplePair<Status, byte[]> readResult) {
      Preconditions.checkState(internalScheduler.isRunningOnThread(), "Not on internal thread");
      final byte[] serializedState = readResult.getFirst().isSuccess() ?
          readResult.getSecond() : null;
      // Call start now.
      if (!readResult.getFirst().isSuccess()) {
        statistics.recordError(ClientErrorType.PERSISTENT_READ_FAILURE);
        logger.warning("Could not read state blob: %s", readResult.getFirst().getMessage());
      }
      startInternal(serializedState);
    }
  });
}
 

import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
@Override  // InvalidationClient
public void acknowledge(final AckHandle acknowledgeHandle) {
  Preconditions.checkNotNull(acknowledgeHandle);
  Preconditions.checkState(internalScheduler.isRunningOnThread(),
      "Not running on internal thread");

  // Parse and validate the ack handle first.
  AckHandleP ackHandle;
  try {
    ackHandle = AckHandleP.parseFrom(acknowledgeHandle.getHandleData());
  } catch (ValidationException exception) {
    logger.warning("Bad ack handle : %s",
        Bytes.toLazyCompactString(acknowledgeHandle.getHandleData()));
    statistics.recordError(ClientErrorType.ACKNOWLEDGE_HANDLE_FAILURE);
    return;
  }

  // Currently, only invalidations have non-trivial ack handle.
  InvalidationP invalidation = ackHandle.getNullableInvalidation();
  if (invalidation == null) {
    logger.warning("Ack handle without invalidation : %s",
        Bytes.toLazyCompactString(acknowledgeHandle.getHandleData()));
    statistics.recordError(ClientErrorType.ACKNOWLEDGE_HANDLE_FAILURE);
    return;
  }

  // Don't send the payload back.
  if (invalidation.hasPayload()) {
    InvalidationP.Builder builder = invalidation.toBuilder();
    builder.payload = null;
    invalidation = builder.build();
  }
  statistics.recordIncomingOperation(IncomingOperationType.ACKNOWLEDGE);
  protocolHandler.sendInvalidationAck(invalidation, batchingTask);

  // Record that the invalidation has been acknowledged to potentially avoid unnecessary delivery
  // of earlier invalidations for the same object.
  ackCache.recordAck(invalidation);
}
 

import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
/**
 * Returns whether the token in the header of {@code parsedMessage} matches either the
 * client token or nonce of this Ticl (depending on which is non-{@code null}).
 */
private boolean validateToken(ParsedMessage parsedMessage) {
  if (clientToken != null) {
    // Client token case.
    if (!TypedUtil.<Bytes>equals(clientToken, parsedMessage.header.token)) {
      logger.info("Incoming message has bad token: server = %s, client = %s",
          parsedMessage.header.token, clientToken);
      statistics.recordError(ClientErrorType.TOKEN_MISMATCH);
      return false;
    }
    return true;
  } else if (nonce != null) {
    // Nonce case.
    if (!TypedUtil.<Bytes>equals(nonce, parsedMessage.header.token)) {
      statistics.recordError(ClientErrorType.NONCE_MISMATCH);
      logger.info("Rejecting server message with mismatched nonce: Client = %s, Server = %s",
          nonce, parsedMessage.header.token);
      return false;
    } else {
      logger.info("Accepting server message with matching nonce: %s", nonce);
      return true;
    }
  }
  // Neither token nor nonce; ignore message.
  logger.warning("Neither token nor nonce was set in validateToken: %s, %s", clientToken, nonce);
  return false;
}
 

import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
/**
 * Handles a message from the server. If the message can be processed (i.e., is valid, is
 * of the right version, and is not a silence message), returns a {@link ParsedMessage}
 * representing it. Otherwise, returns {@code null}.
 * <p>
 * This class intercepts and processes silence messages. In this case, it will discard any other
 * data in the message.
 * <p>
 * Note that this method does <b>not</b> check the session token of any message.
 */
ParsedMessage handleIncomingMessage(byte[] incomingMessage) {
  Preconditions.checkState(internalScheduler.isRunningOnThread(), "Not on internal thread");
  ServerToClientMessage message;
  try {
    message = ServerToClientMessage.parseFrom(incomingMessage);
  } catch (ValidationException exception) {
    statistics.recordError(ClientErrorType.INCOMING_MESSAGE_FAILURE);
    logger.warning("Incoming message is invalid: %s", Bytes.toLazyCompactString(incomingMessage));
    return null;
  }

  // Check the version of the message.
  if (message.getHeader().getProtocolVersion().getVersion().getMajorVersion() !=
      ClientConstants.PROTOCOL_MAJOR_VERSION) {
    statistics.recordError(ClientErrorType.PROTOCOL_VERSION_FAILURE);
    logger.severe("Dropping message with incompatible version: %s", message);
    return null;
  }

  // Check if it is a ConfigChangeMessage which indicates that messages should no longer be
  // sent for a certain duration. Perform this check before the token is even checked.
  if (message.hasConfigChangeMessage()) {
    ConfigChangeMessage configChangeMsg = message.getConfigChangeMessage();
    statistics.recordReceivedMessage(ReceivedMessageType.CONFIG_CHANGE);
    if (configChangeMsg.hasNextMessageDelayMs()) {  // Validator has ensured that it is positive.
      nextMessageSendTimeMs =
          internalScheduler.getCurrentTimeMs() + configChangeMsg.getNextMessageDelayMs();
    }
    return null;  // Ignore all other messages in the envelope.
  }

  lastKnownServerTimeMs = Math.max(lastKnownServerTimeMs, message.getHeader().getServerTimeMs());
  return new ParsedMessage(message);
}
 

import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
/** Sends pending data to the server (e.g., registrations, acks, registration sync messages). */
void sendMessageToServer() {
  Preconditions.checkState(internalScheduler.isRunningOnThread(), "Not on internal thread");
  if (nextMessageSendTimeMs > internalScheduler.getCurrentTimeMs()) {
    logger.warning("In quiet period: not sending message to server: %s > %s",
        nextMessageSendTimeMs, internalScheduler.getCurrentTimeMs());
    return;
  }

  // Create the message from the batcher.
  ClientToServerMessage message;
  try {
    message = batcher.toMessage(createClientHeader(), listener.getClientToken() != null);
    if (message == null) {
      // Happens when we don't have a token and are not sending an initialize message. Logged
      // in batcher.toMessage().
      return;
    }
  } catch (ProtoWrapper.ValidationArgumentException exception) {
    logger.severe("Tried to send invalid message: %s", batcher);
    statistics.recordError(ClientErrorType.OUTGOING_MESSAGE_FAILURE);
    return;
  }
  ++messageId;

  statistics.recordSentMessage(SentMessageType.TOTAL);
  logger.fine("Sending message to server: %s", message);
  network.sendMessage(message.toByteArray());

  // Record that the message was sent. We're invoking the listener directly, rather than
  // scheduling a new work unit to do it. It would be safer to do a schedule, but that's hard to
  // do in Android, we wrote this listener (it's InvalidationClientCore, so we know what it does),
  // and it's the last line of this function.
  listener.handleMessageSent();
}
 

import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
@Override  // InvalidationClient
public void acknowledge(final AckHandle acknowledgeHandle) {
  Preconditions.checkNotNull(acknowledgeHandle);
  Preconditions.checkState(internalScheduler.isRunningOnThread(),
      "Not running on internal thread");

  // 1. Parse the ack handle first.
  AckHandleP ackHandle;
  try {
    ackHandle = AckHandleP.parseFrom(acknowledgeHandle.getHandleData());
  } catch (InvalidProtocolBufferException exception) {
    logger.warning("Bad ack handle : %s",
      CommonProtoStrings2.toLazyCompactString(acknowledgeHandle.getHandleData()));
    statistics.recordError(ClientErrorType.ACKNOWLEDGE_HANDLE_FAILURE);
    return;
  }

  // 2. Validate ack handle - it should have a valid invalidation.
  if (!ackHandle.hasInvalidation() ||
      !msgValidator.isValid(ackHandle.getInvalidation())) {
    logger.warning("Incorrect ack handle data: %s", acknowledgeHandle);
    statistics.recordError(ClientErrorType.ACKNOWLEDGE_HANDLE_FAILURE);
    return;
  }

  // Currently, only invalidations have non-trivial ack handle.
  InvalidationP invalidation = ackHandle.getInvalidation();
  if (invalidation.hasPayload()) {
    // Don't send the payload back.
    invalidation = invalidation.toBuilder().clearPayload().build();
  }
  statistics.recordIncomingOperation(IncomingOperationType.ACKNOWLEDGE);
  protocolHandler.sendInvalidationAck(invalidation, batchingTask);
}
 

import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
/**
 * Returns whether the token in the header of {@code parsedMessage} matches either the
 * client token or nonce of this Ticl (depending on which is non-{@code null}).
 */
private boolean validateToken(ParsedMessage parsedMessage) {
  if (clientToken != null) {
    // Client token case.
    if (!TypedUtil.<ByteString>equals(clientToken, parsedMessage.header.token)) {
      logger.info("Incoming message has bad token: server = %s, client = %s",
          CommonProtoStrings2.toLazyCompactString(parsedMessage.header.token),
          CommonProtoStrings2.toLazyCompactString(clientToken));
      statistics.recordError(ClientErrorType.TOKEN_MISMATCH);
      return false;
    }
    return true;
  } else if (nonce != null) {
    // Nonce case.
    if (!TypedUtil.<ByteString>equals(nonce, parsedMessage.header.token)) {
      statistics.recordError(ClientErrorType.NONCE_MISMATCH);
      logger.info("Rejecting server message with mismatched nonce: Client = %s, Server = %s",
          CommonProtoStrings2.toLazyCompactString(nonce),
          CommonProtoStrings2.toLazyCompactString(parsedMessage.header.token));
      return false;
    } else {
      logger.info("Accepting server message with matching nonce: %s",
          CommonProtoStrings2.toLazyCompactString(nonce));
      return true;
    }
  }
  // Neither token nor nonce; ignore message.
  logger.warning("Neither token nor nonce was set in validateToken: %s, %s", clientToken, nonce);
  return false;
}
 

import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
/** Sends pending data to the server (e.g., registrations, acks, registration sync messages). */
void sendMessageToServer() {
  Preconditions.checkState(internalScheduler.isRunningOnThread(), "Not on internal thread");
  if (nextMessageSendTimeMs > internalScheduler.getCurrentTimeMs()) {
    logger.warning("In quiet period: not sending message to server: %s > %s",
        nextMessageSendTimeMs, internalScheduler.getCurrentTimeMs());
    return;
  }

  // Create the message from the batcher.
  ClientToServerMessage.Builder msgBuilder =
      batcher.toBuilder(listener.getClientToken() != null);
  if (msgBuilder == null) {
    // Happens when we don't have a token and are not sending an initialize message. Logged
    // in batcher.toBuilder().
    return;
  }
  msgBuilder.setHeader(createClientHeader());
  ++messageId;

  // Validate the message and send it.
  ClientToServerMessage message = msgBuilder.build();
  if (!msgValidator.isValid(message)) {
    logger.severe("Tried to send invalid message: %s", message);
    statistics.recordError(ClientErrorType.OUTGOING_MESSAGE_FAILURE);
    return;
  }

  statistics.recordSentMessage(SentMessageType.TOTAL);
  logger.fine("Sending message to server: {0}",
      CommonProtoStrings2.toLazyCompactString(message, true));
  network.sendMessage(message.toByteArray());

  // Record that the message was sent. We're invoking the listener directly, rather than
  // scheduling a new work unit to do it. It would be safer to do a schedule, but that's hard to
  // do in Android, we wrote this listener (it's InvalidationClientCore, so we know what it does),
  // and it's the last line of this function.
  listener.handleMessageSent();
}
 

import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
@Override
public boolean runTask() {
  if (clientToken == null) {
    // We cannot write without a token. We must do this check before creating the
    // PersistentTiclState because newPersistentTiclState cannot handle null tokens.
    return false;
  }

  // Compute the state that we will write if we decide to go ahead with the write.
  final PersistentTiclState state =
      PersistentTiclState.create(clientToken, lastMessageSendTimeMs);
  byte[] serializedState = PersistenceUtils.serializeState(state, digestFn);

  // Decide whether or not to do the write. The decision varies depending on whether or
  // not the channel supports offline delivery. If we decide not to do the write, then
  // that means the in-memory and stored state match semantically, and the train stops.
  if (config.getChannelSupportsOfflineDelivery()) {
    // For offline delivery, we want the entire state to match, since we write the last
    // send time for every message.
    if (state.equals(lastWrittenState.get())) {
      return false;
    }
  } else {
    // If we do not support offline delivery, we avoid writing the state on each message, and
    // we avoid checking the last-sent time (we check only the client token).
    if (TypedUtil.<Bytes>equals(
        state.getClientToken(), lastWrittenState.get().getClientToken())) {
      return false;
    }
  }

  // We decided to do the write.
  storage.writeKey(CLIENT_TOKEN_KEY, serializedState, new Callback<Status>() {
    @Override
    public void accept(Status status) {
      logger.info("Write state completed: %s for %s", status, state);
      Preconditions.checkState(resources.getInternalScheduler().isRunningOnThread());
      if (status.isSuccess()) {
        // Set lastWrittenToken to be the token that was written (NOT clientToken - which
        // could have changed while the write was happening).
        lastWrittenState.set(state);
      } else {
        statistics.recordError(ClientErrorType.PERSISTENT_WRITE_FAILURE);
      }
    }
  });
  return true;  // Reschedule after timeout to make sure that write does happen.
}
 

import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
/**
 * Implementation of {@link #start} on the internal thread with the persistent
 * {@code serializedState} if any. Starts the TICL protocol and makes the TICL ready to receive
 * registrations, invalidations, etc.
 */
private void startInternal(byte[] serializedState) {
  Preconditions.checkState(internalScheduler.isRunningOnThread(), "Not on internal thread");

  // Initialize the session manager using the persisted client token.
  PersistentTiclState persistentState =
      (serializedState == null) ? null : PersistenceUtils.deserializeState(logger,
          serializedState, digestFn);

  if ((serializedState != null) && (persistentState == null)) {
    // In this case, we'll proceed as if we had no persistent state -- i.e., obtain a new client
    // id from the server.
    statistics.recordError(ClientErrorType.PERSISTENT_DESERIALIZATION_FAILURE);
    logger.severe("Failed deserializing persistent state: %s",
        Bytes.toLazyCompactString(serializedState));
  }
  if (persistentState != null) {
    // If we have persistent state, use the previously-stored token and send a heartbeat to
    // let the server know that we've restarted, since we may have been marked offline.

    // In the common case, the server will already have all of our
    // registrations, but we won't know for sure until we've gotten its summary.
    // We'll ask the application for all of its registrations, but to avoid
    // making the registrar redo the work of performing registrations that
    // probably already exist, we'll suppress sending them to the registrar.
    logger.info("Restarting from persistent state: %s", persistentState.getClientToken());
    setNonce(null);
    setClientToken(persistentState.getClientToken());
    shouldSendRegistrations = false;

    // Schedule an info message for the near future.
    int initialHeartbeatDelayMs = computeInitialPersistentHeartbeatDelayMs(
        config, resources, persistentState.getLastMessageSendTimeMs());
    initialPersistentHeartbeatTask = new InitialPersistentHeartbeatTask(initialHeartbeatDelayMs);
    initialPersistentHeartbeatTask.ensureScheduled("");

    // We need to ensure that heartbeats are sent, regardless of whether we start fresh or from
    // persistent state. The line below ensures that they are scheduled in the persistent startup
    // case. For the other case, the task is scheduled when we acquire a token.
    heartbeatTask.ensureScheduled("Startup-after-persistence");
  } else {
    // If we had no persistent state or couldn't deserialize the state that we had, start fresh.
    // Request a new client identifier.

    // The server can't possibly have our registrations, so whatever we get
    // from the application we should send to the registrar.
    logger.info("Starting with no previous state");
    shouldSendRegistrations = true;
    acquireToken("Startup");
  }

  // listener.ready() is called when ticl has acquired a new token.
}
 

import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
/**
 * Handles registration operation statuses from the server. Returns a list of booleans, one per
 * registration status, that indicates whether the registration operation was both successful and
 * agreed with the desired client state (i.e., for each registration status,
 * (status.optype == register) == desiredRegistrations.contains(status.objectid)).
 * <p>
 * REQUIRES: the caller subsequently make an informRegistrationStatus or informRegistrationFailure
 * upcall on the listener for each registration in {@code registrationStatuses}.
 */
List<Boolean> handleRegistrationStatus(List<RegistrationStatus> registrationStatuses) {
  // Local-processing result code for each element of registrationStatuses.
  List<Boolean> localStatuses = new ArrayList<Boolean>(registrationStatuses.size());
  for (RegistrationStatus registrationStatus : registrationStatuses) {
    ObjectIdP objectIdProto = registrationStatus.getRegistration().getObjectId();

    // The object is no longer pending, since we have received a server status for it, so
    // remove it from the pendingOperations map. (It may or may not have existed in the map,
    // since we can receive spontaneous status messages from the server.)
    TypedUtil.remove(pendingOperations, objectIdProto);

    // We start off with the local-processing set as success, then potentially fail.
    boolean isSuccess = true;

    // if the server operation succeeded, then local processing fails on "incompatibility" as
    // defined above.
    if (CommonProtos.isSuccess(registrationStatus.getStatus())) {
      boolean appWantsRegistration = desiredRegistrations.contains(objectIdProto);
      boolean isOpRegistration =
          registrationStatus.getRegistration().getOpType() == RegistrationP.OpType.REGISTER;
      boolean discrepancyExists = isOpRegistration ^ appWantsRegistration;
      if (discrepancyExists) {
        // Remove the registration and set isSuccess to false, which will cause the caller to
        // issue registration-failure to the application.
        desiredRegistrations.remove(objectIdProto);
        statistics.recordError(ClientErrorType.REGISTRATION_DISCREPANCY);
        logger.info("Ticl discrepancy detected: registered = %s, requested = %s. " +
            "Removing %s from requested",
            isOpRegistration, appWantsRegistration, objectIdProto);
        isSuccess = false;
      }
    } else {
      // If the server operation failed, then also local processing fails.
      desiredRegistrations.remove(objectIdProto);
      logger.fine("Removing %s from committed", objectIdProto);
      isSuccess = false;
    }
    localStatuses.add(isSuccess);
  }
  return localStatuses;
}
 

import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
/**
 * Returns a builder for a {@link ClientToServerMessage} to be sent to the server. Crucially,
 * the builder does <b>NOT</b> include the message header.
 * @param hasClientToken whether the client currently holds a token
 */
ClientToServerMessage toMessage(final ClientHeader header, boolean hasClientToken) {
  final InitializeMessage initializeMessage;
  final RegistrationMessage registrationMessage;
  final RegistrationSyncMessage registrationSyncMessage;
  final InvalidationMessage invalidationAckMessage;
  final InfoMessage infoMessage;

  if (pendingInitializeMessage != null) {
    statistics.recordSentMessage(SentMessageType.INITIALIZE);
    initializeMessage = pendingInitializeMessage;
    pendingInitializeMessage = null;
  } else {
    initializeMessage = null;
  }

  // Note: Even if an initialize message is being sent, we can send additional
  // messages such as regisration messages, etc to the server. But if there is no token
  // and an initialize message is not being sent, we cannot send any other message.

  if (!hasClientToken && (initializeMessage == null)) {
    // Cannot send any message
    resources.getLogger().warning(
        "Cannot send message since no token and no initialize msg");
    statistics.recordError(ClientErrorType.TOKEN_MISSING_FAILURE);
    return null;
  }

  // Check for pending batched operations and add to message builder if needed.

  // Add reg, acks, reg subtrees - clear them after adding.
  if (!pendingAckedInvalidations.isEmpty()) {
    invalidationAckMessage = createInvalidationAckMessage();
    statistics.recordSentMessage(SentMessageType.INVALIDATION_ACK);
  } else {
    invalidationAckMessage = null;
  }

  // Check regs.
  if (!pendingRegistrations.isEmpty()) {
    registrationMessage = createRegistrationMessage();
    statistics.recordSentMessage(SentMessageType.REGISTRATION);
  } else {
    registrationMessage = null;
  }

  // Check reg substrees.
  if (!pendingRegSubtrees.isEmpty()) {
    // If there are multiple pending reg subtrees, only one is sent.
    ArrayList<RegistrationSubtree> regSubtrees = new ArrayList<RegistrationSubtree>(1);
    regSubtrees.add(pendingRegSubtrees.iterator().next());
    registrationSyncMessage = RegistrationSyncMessage.create(regSubtrees);
    pendingRegSubtrees.clear();
    statistics.recordSentMessage(SentMessageType.REGISTRATION_SYNC);
  } else {
    registrationSyncMessage = null;
  }

  // Check if an info message has to be sent.
  if (pendingInfoMessage != null) {
    statistics.recordSentMessage(SentMessageType.INFO);
    infoMessage = pendingInfoMessage;
    pendingInfoMessage = null;
  } else {
    infoMessage = null;
  }

  return ClientToServerMessage.create(header, initializeMessage, registrationMessage,
      registrationSyncMessage, invalidationAckMessage, infoMessage);
}
 

import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
@Override
public boolean runTask() {
  if (clientToken == null) {
    // We cannot write without a token. We must do this check before creating the
    // PersistentTiclState because newPersistentTiclState cannot handle null tokens.
    return false;
  }

  // Compute the state that we will write if we decide to go ahead with the write.
  final ProtoWrapper<PersistentTiclState> state =
      ProtoWrapper.of(CommonProtos2.newPersistentTiclState(clientToken, lastMessageSendTimeMs));
  byte[] serializedState = PersistenceUtils.serializeState(state.getProto(), digestFn);

  // Decide whether or not to do the write. The decision varies depending on whether or
  // not the channel supports offline delivery. If we decide not to do the write, then
  // that means the in-memory and stored state match semantically, and the train stops.
  if (config.getChannelSupportsOfflineDelivery()) {
    // For offline delivery, we want the entire state to match, since we write the last
    // send time for every message.
    if (state.equals(lastWrittenState.get())) {
      return false;
    }
  } else {
    // If we do not support offline delivery, we avoid writing the state on each message, and
    // we avoid checking the last-sent time (we check only the client token).
    if (state.getProto().getClientToken().equals(
            lastWrittenState.get().getProto().getClientToken())) {
      return false;
    }
  }

  // We decided to do the write.
  storage.writeKey(CLIENT_TOKEN_KEY, serializedState, new Callback<Status>() {
    @Override
    public void accept(Status status) {
      logger.info("Write state completed: %s for %s", status, state.getProto());
      Preconditions.checkState(resources.getInternalScheduler().isRunningOnThread());
      if (status.isSuccess()) {
        // Set lastWrittenToken to be the token that was written (NOT clientToken - which
        // could have changed while the write was happening).
        lastWrittenState.set(state);
      } else {
        statistics.recordError(ClientErrorType.PERSISTENT_WRITE_FAILURE);
      }
    }
  });
  return true;  // Reschedule after timeout to make sure that write does happen.
}
 

import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
/**
 * Implementation of {@link #start} on the internal thread with the persistent
 * {@code serializedState} if any. Starts the TICL protocol and makes the TICL ready to receive
 * registrations, invalidations, etc.
 */
private void startInternal(byte[] serializedState) {
  Preconditions.checkState(internalScheduler.isRunningOnThread(), "Not on internal thread");

  // Initialize the session manager using the persisted client token.
  PersistentTiclState persistentState =
      (serializedState == null) ? null : PersistenceUtils.deserializeState(logger,
          serializedState, digestFn);

  if ((serializedState != null) && (persistentState == null)) {
    // In this case, we'll proceed as if we had no persistent state -- i.e., obtain a new client
    // id from the server.
    statistics.recordError(ClientErrorType.PERSISTENT_DESERIALIZATION_FAILURE);
    logger.severe("Failed deserializing persistent state: %s",
        CommonProtoStrings2.toLazyCompactString(serializedState));
  }
  if (persistentState != null) {
    // If we have persistent state, use the previously-stored token and send a heartbeat to
    // let the server know that we've restarted, since we may have been marked offline.

    // In the common case, the server will already have all of our
    // registrations, but we won't know for sure until we've gotten its summary.
    // We'll ask the application for all of its registrations, but to avoid
    // making the registrar redo the work of performing registrations that
    // probably already exist, we'll suppress sending them to the registrar.
    logger.info("Restarting from persistent state: %s",
        CommonProtoStrings2.toLazyCompactString(persistentState.getClientToken()));
    setNonce(null);
    setClientToken(persistentState.getClientToken());
    shouldSendRegistrations = false;

    // Schedule an info message for the near future.
    int initialHeartbeatDelayMs = computeInitialPersistentHeartbeatDelayMs(
        config, resources, persistentState.getLastMessageSendTimeMs());
    initialPersistentHeartbeatTask = new InitialPersistentHeartbeatTask(initialHeartbeatDelayMs);
    initialPersistentHeartbeatTask.ensureScheduled("");

    // We need to ensure that heartbeats are sent, regardless of whether we start fresh or from
    // persistent state. The line below ensures that they are scheduled in the persistent startup
    // case. For the other case, the task is scheduled when we acquire a token.
    heartbeatTask.ensureScheduled("Startup-after-persistence");
  } else {
    // If we had no persistent state or couldn't deserialize the state that we had, start fresh.
    // Request a new client identifier.

    // The server can't possibly have our registrations, so whatever we get
    // from the application we should send to the registrar.
    logger.info("Starting with no previous state");
    shouldSendRegistrations = true;
    acquireToken("Startup");
  }

  // listener.ready() is called when ticl has acquired a new token.
}
 

import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
/**
 * Handles registration operation statuses from the server. Returns a list of booleans, one per
 * registration status, that indicates whether the registration operation was both successful and
 * agreed with the desired client state (i.e., for each registration status,
 * (status.optype == register) == desiredRegistrations.contains(status.objectid)).
 * <p>
 * REQUIRES: the caller subsequently make an informRegistrationStatus or informRegistrationFailure
 * upcall on the listener for each registration in {@code registrationStatuses}.
 */
List<Boolean> handleRegistrationStatus(List<RegistrationStatus> registrationStatuses) {
  // Local-processing result code for each element of registrationStatuses.
  List<Boolean> localStatuses = new ArrayList<Boolean>(registrationStatuses.size());
  for (RegistrationStatus registrationStatus : registrationStatuses) {
    ObjectIdP objectIdProto = registrationStatus.getRegistration().getObjectId();

    // The object is no longer pending, since we have received a server status for it, so
    // remove it from the pendingOperations map. (It may or may not have existed in the map,
    // since we can receive spontaneous status messages from the server.)
    TypedUtil.remove(pendingOperations, ProtoWrapper.of(objectIdProto));

    // We start off with the local-processing set as success, then potentially fail.
    boolean isSuccess = true;

    // if the server operation succeeded, then local processing fails on "incompatibility" as
    // defined above.
    if (CommonProtos2.isSuccess(registrationStatus.getStatus())) {
      boolean appWantsRegistration = desiredRegistrations.contains(objectIdProto);
      boolean isOpRegistration =
          registrationStatus.getRegistration().getOpType() == RegistrationP.OpType.REGISTER;
      boolean discrepancyExists = isOpRegistration ^ appWantsRegistration;
      if (discrepancyExists) {
        // Remove the registration and set isSuccess to false, which will cause the caller to
        // issue registration-failure to the application.
        desiredRegistrations.remove(objectIdProto);
        statistics.recordError(ClientErrorType.REGISTRATION_DISCREPANCY);
        logger.info("Ticl discrepancy detected: registered = %s, requested = %s. " +
            "Removing %s from requested",
            isOpRegistration, appWantsRegistration,
            CommonProtoStrings2.toLazyCompactString(objectIdProto));
        isSuccess = false;
      }
    } else {
      // If the server operation failed, then also local processing fails.
      desiredRegistrations.remove(objectIdProto);
      logger.fine("Removing %s from committed",
          CommonProtoStrings2.toLazyCompactString(objectIdProto));
      isSuccess = false;
    }
    localStatuses.add(isSuccess);
  }
  return localStatuses;
}
 

import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
/**
 * Returns a builder for a {@link ClientToServerMessage} to be sent to the server. Crucially,
 * the builder does <b>NOT</b> include the message header.
 * @param hasClientToken whether the client currently holds a token
 */
ClientToServerMessage.Builder toBuilder(boolean hasClientToken) {
  ClientToServerMessage.Builder builder = ClientToServerMessage.newBuilder();
  if (pendingInitializeMessage != null) {
    statistics.recordSentMessage(SentMessageType.INITIALIZE);
    builder.setInitializeMessage(pendingInitializeMessage);
    pendingInitializeMessage = null;
  }

  // Note: Even if an initialize message is being sent, we can send additional
  // messages such as regisration messages, etc to the server. But if there is no token
  // and an initialize message is not being sent, we cannot send any other message.

  if (!hasClientToken && !builder.hasInitializeMessage()) {
    // Cannot send any message
    resources.getLogger().warning(
        "Cannot send message since no token and no initialize msg: %s", builder);
    statistics.recordError(ClientErrorType.TOKEN_MISSING_FAILURE);
    return null;
  }

  // Check for pending batched operations and add to message builder if needed.

  // Add reg, acks, reg subtrees - clear them after adding.
  if (!pendingAckedInvalidations.isEmpty()) {
    builder.setInvalidationAckMessage(createInvalidationAckMessage());
    statistics.recordSentMessage(SentMessageType.INVALIDATION_ACK);
  }

  // Check regs.
  if (!pendingRegistrations.isEmpty()) {
    builder.setRegistrationMessage(createRegistrationMessage());
    statistics.recordSentMessage(SentMessageType.REGISTRATION);
  }

  // Check reg substrees.
  if (!pendingRegSubtrees.isEmpty()) {
    for (ProtoWrapper<RegistrationSubtree> subtree : pendingRegSubtrees) {
      builder.setRegistrationSyncMessage(RegistrationSyncMessage.newBuilder()
          .addSubtree(subtree.getProto()));
    }
    pendingRegSubtrees.clear();
    statistics.recordSentMessage(SentMessageType.REGISTRATION_SYNC);
  }

  // Check if an info message has to be sent.
  if (pendingInfoMessage != null) {
    statistics.recordSentMessage(SentMessageType.INFO);
    builder.setInfoMessage(pendingInfoMessage);
    pendingInfoMessage = null;
  }
  return builder;
}
 

import com.google.ipc.invalidation.ticl.Statistics.ClientErrorType; //导入依赖的package包/类
/**
 * Handles a message from the server. If the message can be processed (i.e., is valid, is
 * of the right version, and is not a silence message), returns a {@link ParsedMessage}
 * representing it. Otherwise, returns {@code null}.
 * <p>
 * This class intercepts and processes silence messages. In this case, it will discard any other
 * data in the message.
 * <p>
 * Note that this method does <b>not</b> check the session token of any message.
 */
ParsedMessage handleIncomingMessage(byte[] incomingMessage) {
  Preconditions.checkState(internalScheduler.isRunningOnThread(), "Not on internal thread");
  ServerToClientMessage message;
  try {
    message = ServerToClientMessage.parseFrom(incomingMessage);
  } catch (InvalidProtocolBufferException exception) {
    logger.warning("Incoming message is unparseable: %s",
        CommonProtoStrings2.toLazyCompactString(incomingMessage));
    return null;
  }

  // Validate the message. If this passes, we can blindly assume valid messages from here on.
  logger.fine("Incoming message: %s", message);
  if (!msgValidator.isValid(message)) {
    statistics.recordError(ClientErrorType.INCOMING_MESSAGE_FAILURE);
    logger.severe("Received invalid message: %s", message);
    return null;
  }

  // Check the version of the message.
  if (message.getHeader().getProtocolVersion().getVersion().getMajorVersion() !=
      CommonInvalidationConstants2.PROTOCOL_MAJOR_VERSION) {
    statistics.recordError(ClientErrorType.PROTOCOL_VERSION_FAILURE);
    logger.severe("Dropping message with incompatible version: %s", message);
    return null;
  }

  // Check if it is a ConfigChangeMessage which indicates that messages should no longer be
  // sent for a certain duration. Perform this check before the token is even checked.
  if (message.hasConfigChangeMessage()) {
    ConfigChangeMessage configChangeMsg = message.getConfigChangeMessage();
    statistics.recordReceivedMessage(ReceivedMessageType.CONFIG_CHANGE);
    if (configChangeMsg.hasNextMessageDelayMs()) {  // Validator has ensured that it is positive.
      nextMessageSendTimeMs =
          internalScheduler.getCurrentTimeMs() + configChangeMsg.getNextMessageDelayMs();
    }
    return null;  // Ignore all other messages in the envelope.
  }

  lastKnownServerTimeMs = Math.max(lastKnownServerTimeMs, message.getHeader().getServerTimeMs());
  return new ParsedMessage(message);
}