Creates a DIFF message.

@param lastZxid the last zxid of the server who initiates the sync.
@return a protobuf message.
180133
public static Message buildDiff(Zxid lastZxid) {
    ZabMessage.Zxid lz = toProtoZxid(lastZxid);

    Diff diff = Diff.newBuilder().setLastZxid(lz).build();

    return Message.newBuilder().setType(MessageType.DIFF)
                               .setDiff(diff)
                               .build();
  }
926486
Zxid getZxidFromFileName(File file) {
    String fileName = file.getName();
    String strZxid = fileName.substring(fileName.indexOf('.') + 1);
    return Zxid.fromSimpleString(strZxid);
  }
1338803
public ZMsg duplicate()
    {
        if (frames.isEmpty()) {
            return null;
        }
        else {
            ZMsg msg = new ZMsg();
            for (ZFrame f : frames) {
                msg.add(f.duplicate());
            }
            return msg;
        }
    }
284106
public BaseMessage getOriginalMessage(BaseMessage messageResponseIn)
    {
        BaseMessage messageOrig = null;
        TrxMessageHeader trxMessageHeader = (TrxMessageHeader)messageResponseIn.getMessageHeader();
        String strOrigTrxID = (String)trxMessageHeader.get(TrxMessageHeader.ORIG_LOG_TRX_ID);
        MessageLogModel recMessageLog = this.getMessageLog(strOrigTrxID);
        if ((recMessageLog != null)
            && ((recMessageLog.getEditMode() == DBConstants.EDIT_CURRENT)
                || (recMessageLog.getEditMode() == DBConstants.EDIT_IN_PROGRESS)))
        {
            messageOrig = (BaseMessage)recMessageLog.createMessage(strOrigTrxID);
        }
        return messageOrig;
    }
348131
public final ZEvent nextEvent(boolean wait)
    {
        if (!started) {
            System.out.println("ZMonitor: Start before getting events.");
            return null;
        }
        ZMsg msg = agent.recv(wait);
        if (msg == null) {
            return null;
        }
        return new ZEvent(msg);
    }
718964
private void joinSynchronization()
      throws IOException, TimeoutException, InterruptedException {
    Message sync = MessageBuilder.buildSyncHistory(persistence.getLatestZxid());
    sendMessage(this.electedLeader, sync);
    MessageTuple tuple =
      filter.getExpectedMessage(MessageType.SYNC_HISTORY_REPLY, electedLeader,
                                config.getTimeoutMs());
    Message msg = tuple.getMessage();
    // Updates the sync timeout based on leader's suggestion.
    setSyncTimeoutMs(msg.getSyncHistoryReply().getSyncTimeout());
    // Waits for the first synchronization completes.
    waitForSync(this.electedLeader);

    // Gets the last zxid in disk after first synchronization.
    Zxid lastZxid = persistence.getLatestZxid();
    LOG.debug("After first synchronization, the last zxid is {}", lastZxid);
    // Then issues the JOIN message.
    Message join = MessageBuilder.buildJoin(lastZxid);
    sendMessage(this.electedLeader, join);

    /* -- Synchronizing phase -- */
    changePhase(Phase.SYNCHRONIZING);
    waitForSync(this.electedLeader);
    waitForNewLeaderMessage();
    waitForCommitMessage();
    persistence.setProposedEpoch(persistence.getAckEpoch());
  }
448273
void synchronizeFromFollower(String peerId)
      throws IOException, TimeoutException, InterruptedException {
    LOG.debug("Begins synchronizing from follower {}.", peerId);
    Zxid lastZxid = persistence.getLatestZxid();
    Message pullTxn = MessageBuilder.buildPullTxnReq(lastZxid);
    LOG.debug("Last zxid of {} is {}", this.serverId, lastZxid);
    sendMessage(peerId, pullTxn);
    // Waits until the synchronization is finished.
    waitForSync(peerId);
  }
128434
public int diff_xIndex(LinkedList<Diff> diffs, int loc) {
        int chars1 = 0;
        int chars2 = 0;
        int last_chars1 = 0;
        int last_chars2 = 0;
        Diff lastDiff = null;
        for (Diff aDiff : diffs) {
            if (aDiff.operation != Operation.INSERT) {
                // Equality or deletion.
                chars1 += aDiff.text.length();
            }
            if (aDiff.operation != Operation.DELETE) {
                // Equality or insertion.
                chars2 += aDiff.text.length();
            }
            if (chars1 > loc) {
                // Overshot the location.
                lastDiff = aDiff;
                break;
            }
            last_chars1 = chars1;
            last_chars2 = chars2;
        }
        if (lastDiff != null && lastDiff.operation == Operation.DELETE) {
            // The location was deleted.
            return last_chars2;
        }
        // Add the remaining character length.
        return last_chars2 + (loc - last_chars1);
    }
968864
public ZookeeperTxId getPrimaryLastTxId(String address, boolean sync)
      throws IOException,
      KeeperException, InterruptedException, ClassNotFoundException {
    Stat stat = new Stat();
    String node = getLastTxIdNode(address);
    byte[] data = getNodeData(node, stat, false, sync);
    if (data == null) {
      return null;
    }
    return ZookeeperTxId.getFromBytes(data);
  }
1346077
@SuppressWarnings("unchecked")
    public static UniversalIdStrQueueMessage newInstance() {
        Date now = new Date();
        UniversalIdStrQueueMessage msg = new UniversalIdStrQueueMessage();
        msg.setQueueTimestamp(now).setNumRequeues(0)
                .setId(QueueUtils.IDGEN.generateId128Hex().toLowerCase()).setTimestamp(now);
        return msg;
    }
1015917
public static ZabMessage.Zxid toProtoZxid(Zxid zxid) {
    ZabMessage.Zxid z = ZabMessage.Zxid.newBuilder()
                                  .setEpoch(zxid.getEpoch())
                                  .setXid(zxid.getXid())
                                  .build();
    return z;
  }
749379
@Local
    public CompletableFuture<Integer> sendMessage(final Convert convert, final Object message0, final boolean last, final Serializable... userids) {
        if (userids == null || userids.length < 1) return CompletableFuture.completedFuture(RETCODE_GROUP_EMPTY);
        if (message0 instanceof CompletableFuture) return ((CompletableFuture) message0).thenApply(msg -> sendMessage(convert, msg, last, userids));
        final Object message = (convert == null || message0 instanceof WebSocketPacket) ? message0 : ((convert instanceof TextConvert) ? new WebSocketPacket(((TextConvert) convert).convertTo(message0), last) : new WebSocketPacket(((BinaryConvert) convert).convertTo(message0), last));
        if (this.localEngine != null && this.sncpNodeAddresses == null) { //本地模式且没有分布式
            return this.localEngine.sendLocalMessage(message, last, userids);
        }
        final Object remoteMessage = formatRemoteMessage(message);
        CompletableFuture<Integer> rsfuture;
        if (userids.length == 1) {
            rsfuture = sendOneUserMessage(remoteMessage, last, userids[0]);
        } else {
            String[] keys = new String[userids.length];
            final Map<String, Serializable> keyuser = new HashMap<>();
            for (int i = 0; i < userids.length; i++) {
                keys[i] = SOURCE_SNCP_USERID_PREFIX + userids[i];
                keyuser.put(keys[i], userids[i]);
            }
            tryAcquireSemaphore();
            CompletableFuture<Map<String, Collection<InetSocketAddress>>> addrsFuture = sncpNodeAddresses.getCollectionMapAsync(InetSocketAddress.class, keys);
            if (semaphore != null) addrsFuture.whenComplete((r, e) -> releaseSemaphore());
            rsfuture = addrsFuture.thenCompose((Map<String, Collection<InetSocketAddress>> addrs) -> {
                if (addrs == null || addrs.isEmpty()) {
                    if (logger.isLoggable(Level.FINER)) logger.finer("websocket not found userids:" + JsonConvert.root().convertTo(userids) + " on any node ");
                    return CompletableFuture.completedFuture(RETCODE_GROUP_EMPTY);
                }
                Map<InetSocketAddress, List<Serializable>> addrUsers = new HashMap<>();
                addrs.forEach((key, as) -> {
                    for (InetSocketAddress a : as) {
                        addrUsers.computeIfAbsent(a, k -> new ArrayList<>()).add(keyuser.get(key));
                    }
                });
                if (logger.isLoggable(Level.FINEST)) {
                    logger.finest("websocket(localaddr=" + localSncpAddress + ", userids=" + JsonConvert.root().convertTo(userids) + ") found message-addr-userids: " + addrUsers);
                }
                CompletableFuture<Integer> future = null;
                for (Map.Entry<InetSocketAddress, List<Serializable>> en : addrUsers.entrySet()) {
                    Serializable[] us = en.getValue().toArray(new Serializable[en.getValue().size()]);
                    future = future == null ? sendOneAddrMessage(en.getKey(), remoteMessage, last, us)
                        : future.thenCombine(sendOneAddrMessage(en.getKey(), remoteMessage, last, us), (a, b) -> a | b);
                }
                return future == null ? CompletableFuture.completedFuture(RETCODE_GROUP_EMPTY) : future;
            });
        }
        return rsfuture == null ? CompletableFuture.completedFuture(RETCODE_GROUP_EMPTY) : rsfuture;
    }
440589
def createZMQSocket(self, sock_type):
        """"""
        sock = self.ZMQcontext.socket(sock_type)
        sock.setsockopt(zmq.LINGER, LINGER_TIME)
        sock.setsockopt(zmq.IPV4ONLY, 0)

        # Remove message dropping
        sock.setsockopt(zmq.SNDHWM, 0)
        sock.setsockopt(zmq.RCVHWM, 0)
        try:
            sock.setsockopt(zmq.IMMEDIATE, 1)
        except:
            # This parameter was recently added by new libzmq versions
            pass

        # Don't accept unroutable messages
        if sock_type == zmq.ROUTER:
            sock.setsockopt(zmq.ROUTER_MANDATORY, 1)
        return sock
1319973
public static Message buildAck(Zxid zxid) {
    ZabMessage.Zxid zzxid = toProtoZxid(zxid);
    Ack ack = Ack.newBuilder().setZxid(zzxid).build();
    return Message.newBuilder().setType(MessageType.ACK).setAck(ack).build();
  }
13386
public static Message buildDelivered(Zxid deliveredZxid) {
    ZabMessage.Zxid zxid = toProtoZxid(deliveredZxid);
    ZabMessage.Delivered delivered =
      ZabMessage.Delivered.newBuilder().setZxid(zxid).build();
    return Message.newBuilder().setType(MessageType.DELIVERED)
                               .setDelivered(delivered).build();
  }
200918
def send_last_message(self, msg, connection_id=None):
        
        zmq_identity = None
        if connection_id is not None and self._connections is not None:
            if connection_id in self._connections:
                connection_info = self._connections.get(connection_id)
                if connection_info.connection_type == \
                        ConnectionType.ZMQ_IDENTITY:
                    zmq_identity = connection_info.connection
                del self._connections[connection_id]

            else:
                LOGGER.debug("Can't send to %s, not in self._connections",
                             connection_id)
                return

        self._ready.wait()

        try:
            asyncio.run_coroutine_threadsafe(
                self._send_last_message(zmq_identity, msg),
                self._event_loop)
        except RuntimeError:
            # run_coroutine_threadsafe will throw a RuntimeError if
            # the eventloop is closed. This occurs on shutdown.
            pass
567970
void restoreFromSnapshot() throws IOException {
    Zxid snapshotZxid = persistence.getSnapshotZxid();
    LOG.debug("The last applied zxid in snapshot is {}", snapshotZxid);
    if (snapshotZxid != Zxid.ZXID_NOT_EXIST &&
        lastDeliveredZxid == Zxid.ZXID_NOT_EXIST) {
      // Restores from snapshot only if the lastDeliveredZxid == ZXID_NOT_EXIST,
      // which means it's the application's first time recovery.
      File snapshot = persistence.getSnapshotFile();
      try (FileInputStream fin = new FileInputStream(snapshot)) {
        LOG.debug("Restoring application's state from snapshot {}", snapshot);
        stateMachine.restore(fin);
        lastDeliveredZxid = snapshotZxid;
      }
    }
  }