Deliver the given {@link SMTPDeliveryEnvelope}'s

The implementation may choose to do the deliver in an async fashion.

@param host
@param config
@param transactions
@return future
506721
public SMTPClientFuture<Collection<FutureResult<Iterator<DeliveryRecipientStatus>>>> deliver(InetSocketAddress host, final SMTPDeliveryAgentConfig config, final SMTPDeliveryEnvelope... transactions) {
        if (transactions == null || transactions.length == 0) {
            throw new IllegalArgumentException("SMTPTransaction parameter must be not null and the length must be > 0");
        }

        final SMTPClientFutureImpl<Collection<FutureResult<Iterator<DeliveryRecipientStatus>>>> future = new SMTPClientFutureImpl<Collection<FutureResult<Iterator<DeliveryRecipientStatus>>>>();
        final SMTPClientFutureListenerFactory factory = createFactory();
        
        transport.connect(host, config).addListener(new ChainedSMTPClientFutureListener<SMTPResponse>() {
          

            @Override
            public void onException(SMTPClientSession session, SMTPException t) {
                initSession(session);
                super.onException(session, t);
            
            }
           
            /**
             * Init the SMTPClienSesion by adding all needed data to the attributes
             * 
             * @param session
             */
            private void initSession(SMTPClientSession session) {
                
                
                Iterator<SMTPDeliveryEnvelope> transactionIt = new ArrayIterator<SMTPDeliveryEnvelope>(transactions);
                session.setAttribute(SMTP_TRANSACTIONS_KEY, transactionIt);
                SMTPDeliveryEnvelope transaction = transactionIt.next();
                session.setAttribute(CURRENT_SMTP_TRANSACTION_KEY, transaction);
                session.setAttribute(RECIPIENTS_KEY, transaction.getRecipients().iterator());

                session.setAttribute(FUTURE_KEY, future);
                session.setAttribute(DELIVERY_STATUS_KEY, new ArrayList<DeliveryRecipientStatus>());
                session.setAttribute(DELIVERY_RESULT_LIST_KEY, new ArrayList<FutureResult<Iterator<DeliveryRecipientStatus>>>());
                session.setAttribute(SMTP_CLIENT_FUTURE_LISTENER_FACTORY, createFactory());
                session.setAttribute(SMTP_TRANSACTION_ACTIVE_KEY, false);
            }

            @Override
            protected void onResult(SMTPClientSession session, SMTPResponse result) throws SMTPException {
                initSession(session);
                FutureResult<SMTPResponse> fResult = new FutureResultImpl<SMTPResponse>(result);
                ReadySMTPClientFuture<FutureResult<SMTPResponse>> future = new ReadySMTPClientFuture<FutureResult<SMTPResponse>>(session, fResult);
                future.addListener(factory.getListener(session));                
            }
        });
        
       
        return future;
    }
1339315
public void deliver(VoltMessage msg)
    {
        if (!m_isLeader && msg instanceof Iv2InitiateTaskMessage) {
            final Iv2InitiateTaskMessage m = (Iv2InitiateTaskMessage) msg;
            // We can't repair read only SP transactions. Just don't log them to the repair log.
            if (m.isReadOnly()) {
                return;
            }
            m_lastSpHandle = m.getSpHandle();
            truncate(m.getTruncationHandle(), IS_SP);

            //Cann't repair MigratePartitionLeader
            if ("@MigratePartitionLeader".equalsIgnoreCase(m.getStoredProcedureName())) {
                return;
            }

            m_logSP.add(new Item(IS_SP, m, m.getSpHandle(), m.getTxnId()));
        } else if (msg instanceof FragmentTaskMessage) {
            boolean newMp = false;
            final FragmentTaskMessage m = (FragmentTaskMessage) msg;
            if (m.getTxnId() > m_lastMpHandle || m_lastMpHandle == Long.MAX_VALUE) {
                m_lastMpHandle = m.getTxnId();
                newMp = true;
            }

            // We can't repair read only MP transactions. Just don't log them to the repair log.
            if (m.isReadOnly()) {
                return;
            }

            truncate(m.getTruncationHandle(), IS_MP);
            // only log the first fragment of a procedure (and handle 1st case)
            if (newMp) {
                m_logMP.add(new Item(IS_MP, m, m.getSpHandle(), m.getTxnId()));
                m_lastSpHandle = m.getSpHandle();
            }
        }
        else if (msg instanceof CompleteTransactionMessage) {
            // a CompleteTransactionMessage which indicates restart is not the end of the
            // transaction.  We don't want to log it in the repair log.
            CompleteTransactionMessage ctm = (CompleteTransactionMessage)msg;

            //Restore will send a complete transaction message with a lower mp transaction id because
            //the restore transaction precedes the loading of the right mp transaction id from the snapshot
            //Hence Math.max
            m_lastMpHandle = Math.max(m_lastMpHandle, ctm.getTxnId());

            // We can't repair read only MP transactions. Just don't log them to the repair log.
            // Restart transaction do not need to be repaired here, don't log them as well.
            if (ctm.isReadOnly() || ctm.isRestart() || ctm.isAbortDuringRepair()) {
                return;
            }

            truncate(ctm.getTruncationHandle(), IS_MP);
            m_logMP.add(new Item(IS_MP, ctm, ctm.getSpHandle(), ctm.getTxnId()));
            m_lastSpHandle = ctm.getSpHandle();
        } else if (msg instanceof DumpMessage) {
            String who = CoreUtils.hsIdToString(m_HSId);
            repairLogger.warn("Repair log dump for site: " + who + ", isLeader: " + m_isLeader
                    + ", " + who + ": lastSpHandle: " + m_lastSpHandle + ", lastMpHandle: " + m_lastMpHandle);
            for (Iv2RepairLogResponseMessage il : contents(0l, false)) {
               repairLogger.warn("[Repair log contents]" + who + ": msg: " + il);
            }
        } else if (msg instanceof DummyTransactionTaskMessage) {
            m_lastSpHandle = Math.max(m_lastSpHandle, ((DummyTransactionTaskMessage) msg).getSpHandle());
        } else if (msg instanceof RepairLogTruncationMessage) {
            final RepairLogTruncationMessage truncateMsg = (RepairLogTruncationMessage) msg;
            truncate(truncateMsg.getHandle(), IS_SP);
        }
    }
1332906
public byte[] deliverSm(int sequenceNumber, String serviceType,
            byte sourceAddrTon, byte sourceAddrNpi, String sourceAddr,
            byte destAddrTon, byte destAddrNpi, String destinationAddr,
            byte esmClass, byte protocolId, byte priorityFlag,
            byte registeredDelivery, byte dataCoding, byte[] shortMessage,
            OptionalParameter... optionalParameters) throws PDUStringException {

        StringValidator.validateString(serviceType,
                StringParameter.SERVICE_TYPE);
        StringValidator.validateString(sourceAddr, StringParameter.SOURCE_ADDR);
        StringValidator.validateString(destinationAddr,
                StringParameter.DESTINATION_ADDR);
        StringValidator.validateString(shortMessage,
                StringParameter.SHORT_MESSAGE);

        PDUByteBuffer buf = new PDUByteBuffer(SMPPConstant.CID_DELIVER_SM, 0,
                sequenceNumber);
        buf.append(serviceType);
        buf.append(sourceAddrTon);
        buf.append(sourceAddrNpi);
        buf.append(sourceAddr);
        buf.append(destAddrTon);
        buf.append(destAddrNpi);
        buf.append(destinationAddr);
        buf.append(esmClass);
        buf.append(protocolId);
        buf.append(priorityFlag);
        buf.append((String)null); // schedule delivery time
        buf.append((String)null); // validity period
        buf.append(registeredDelivery);
        buf.append((byte)0); // replace if present flag
        buf.append(dataCoding);
        buf.append((byte)0); // sm default msg id
        buf.append((byte)shortMessage.length);
        buf.append(shortMessage);
        buf.appendAll(optionalParameters);
        ;
        return buf.toBytes();
    }
1040661
@Override
  public QueueDeliver<M> build(Deliver<M> deliver)
  {
    validateBuilder();
    
    if (deliver == null) {
      throw new IllegalArgumentException(L.l("'processors' is required"));
    }
    
    QueueRing<M> queue = buildQueue();
    
    Executor executor = createExecutor();
    ClassLoader loader = getClassLoader();
    
    // OutboxDeliver<M> outbox = _outboxFactory.createOutbox(deliver);
    
    WorkerDeliverSingleThread<M> worker
      = new WorkerDeliverSingleThread<M>(deliver,
                                        _outboxContext,
                                        executor,
                                        loader,
                                        queue);
    
    return new QueueDeliverImpl<M>(queue, worker);
  }
231237
public static ThriftClient build(String host, int port) throws TException {
        return build(host, port, null);
    }
1373078
public ConfigBuilder withHost(final String host) {
        if (host == null || "".equals(host)) {
            throw new IllegalArgumentException("host must not be null or empty: " + host);
        }
        this.host = host;
        return this;
    }
1117614
@SuppressWarnings("deprecation")
	public static MailerRegularBuilderImpl withSMTPServer(@Nullable final String host, @Nullable final Integer port, @Nullable final String username, @Nullable final String password) {
		return new MailerRegularBuilderImpl().withSMTPServer(host, port, username, password);
	}
1183013
protected boolean canDeliver(Address sender, long seqno) {
        BoundedHashMap<Long,Long> seqno_set=delivery_table.get(sender);
        if(seqno_set == null) {
            seqno_set=new BoundedHashMap<>(delivery_table_max_size);
            BoundedHashMap<Long,Long> existing=delivery_table.put(sender,seqno_set);
            if(existing != null)
                seqno_set=existing;
        }
        return seqno_set.add(seqno, seqno);
    }
1110526
public NimbusClient createClient(String host, int port)
    {
        Map<String, Object> nimbusConf = Maps.newHashMap();
        nimbusConf.put(Config.NIMBUS_HOST, host);
        nimbusConf.put(Config.NIMBUS_THRIFT_PORT, port);
        nimbusConf.put(Config.STORM_THRIFT_TRANSPORT_PLUGIN,
                "backtype.storm.security.auth.SimpleTransportPlugin");
        NimbusClient nimbusClient = NimbusClient.getConfiguredClient(nimbusConf);
        return nimbusClient;
    }
1016377
@Override
  public void deliver(WriteStream os, OutHttp2 outHttp)
    throws IOException
  {
    outHttp.writeFlow(_streamId, _credit);
  }
622454
public <T> T withTransaction(@NotNull Propagation propagation,
                                 @NotNull Isolation isolation,
                                 @NotNull TransactionCallback<T> callback) {

        TransactionSettings settings = new TransactionSettings();
        settings.setPropagation(propagation);
        settings.setIsolation(isolation);

        return withTransaction(settings, callback);
    }
1131176
@SuppressWarnings("unchecked")
  private CompletableFuture<Void> completeTransaction(
      TransactionId transactionId,
      TransactionState expectState,
      Function<TransactionInfo, TransactionInfo> updateFunction,
      Predicate<TransactionInfo> updatedPredicate,
      BiFunction<TransactionId, Transactional<?>, CompletableFuture<Void>> completionFunction) {
    return transactions.compute(transactionId, (id, info) -> {
      if (info == null) {
        return null;
      } else if (info.state == expectState) {
        return updateFunction.apply(info);
      } else {
        return info;
      }
    }).thenCompose(value -> {
      if (value != null && updatedPredicate.test(value.value())) {
        return Futures.allOf(value.value().participants.stream()
            .map(participantInfo -> completeParticipant(participantInfo, info -> completionFunction.apply(transactionId, info))))
            .thenApply(v -> null);
      }
      return Futures.exceptionalFuture(new TransactionException("Failed to acquire transaction lock"));
    });
  }
1232979
public OperationFuture<Server> addSecondaryNetwork(Server server, AddNetworkConfig config) {
        ServerMetadata serverMetadata = findByRef(server);

        NetworkLink link = client.addSecondaryNetwork(
            idByRef(server),
            buildSecondaryNetworkRequest(config, serverMetadata.getLocationId())
        );

        return new OperationFuture<>(
            server,
            link.getOperationId(),
            experimentalQueueClient
        );
    }
1257022
void connectEnvelope(Envelope envelope) throws TransactionException {

        if (shutdown) {
            logger.warn(format("%s not connecting is shutdown.", toString()));
            return;
        }

        final AtomicBoolean retry = new AtomicBoolean(true); // make sure we only retry connection once for each connection attempt.

        ManagedChannel lmanagedChannel = managedChannel;

        if (lmanagedChannel == null || lmanagedChannel.isTerminated() || lmanagedChannel.isShutdown()) {

            lmanagedChannel = channelBuilder.build();
            managedChannel = lmanagedChannel;

        }

        try {

            DeliverGrpc.DeliverStub broadcast = DeliverGrpc.newStub(lmanagedChannel);

            // final DeliverResponse[] ret = new DeliverResponse[1];
            //   final List<DeliverResponse> retList = new ArrayList<>();
            final List<Throwable> throwableList = new ArrayList<>();
            final CountDownLatch finishLatch = new CountDownLatch(1);

            so = new StreamObserver<DeliverResponse>() {

                @Override
                public void onNext(DeliverResponse resp) {

                    // logger.info("Got Broadcast response: " + resp);
                    logger.trace(format("DeliverResponse %s resp status value:%d  status %s, typecase %s ",
                            PeerEventServiceClient.this.toString(), resp.getStatusValue(), resp.getStatus(), resp.getTypeCase()));

                    final DeliverResponse.TypeCase typeCase = resp.getTypeCase();

                    if (typeCase == STATUS) {

                        logger.debug(format("DeliverResponse  %s setting done.",
                                PeerEventServiceClient.this.toString()));

                        if (resp.getStatus() == Common.Status.SUCCESS) { // unlike you may think this only happens when all blocks are fetched.
                            peer.setLastConnectTime(System.currentTimeMillis());
                            peer.resetReconnectCount();
                        } else {

                            final long rec = peer.getReconnectCount();

                            PeerEventingServiceException peerEventingServiceException = new PeerEventingServiceException(format("%s attempts %s Status returned failure code %d (%s) during peer service event registration",
                                    PeerEventServiceClient.this.toString(), rec, resp.getStatusValue(), resp.getStatus().name()));
                            peerEventingServiceException.setResponse(resp);
                            if (rec % 10 == 0) {
                                logger.warn(PeerEventServiceClient.this.toString() + " " + peerEventingServiceException.getMessage());
                            }

                            throwableList.add(peerEventingServiceException);
                        }

                    } else if (typeCase == FILTERED_BLOCK || typeCase == BLOCK) {
                        if (typeCase == BLOCK) {
                            logger.trace(format("%s got event block hex hashcode: %016x, block number: %d",
                                    PeerEventServiceClient.this.toString(), resp.getBlock().hashCode(), resp.getBlock().getHeader().getNumber()));
                        } else {
                            logger.trace(format("%s got event block hex hashcode: %016x, block number: %d",
                                    PeerEventServiceClient.this.toString(), resp.getFilteredBlock().hashCode(), resp.getFilteredBlock().getNumber()));
                        }

                        peer.setLastConnectTime(System.currentTimeMillis());
                        long reconnectCount = peer.getReconnectCount();
                        if (reconnectCount > 1) {
                            logger.info(format("%s reconnected after %d attempts on channel %s, peer %s, url %s",
                                    PeerEventServiceClient.this.toString(), reconnectCount, channelName, name, url));
                        }
                        peer.resetReconnectCount();

                        BlockEvent blockEvent = new BlockEvent(peer, resp);
                        peer.setLastBlockSeen(blockEvent);

                        channelEventQue.addBEvent(blockEvent);
                    } else {
                        logger.error(format("%s got event block with unknown type: %s, %d",
                                PeerEventServiceClient.this.toString(), typeCase.name(), typeCase.getNumber()));

                        PeerEventingServiceException peerEventingServiceException = new PeerEventingServiceException(format("% got event block with unknown type: %s, %d",
                                PeerEventServiceClient.this.toString(), typeCase.name(), typeCase.getNumber()));
                        peerEventingServiceException.setResponse(resp);

                        throwableList.add(peerEventingServiceException);

                    }
                    finishLatch.countDown();

                }

                @Override
                public void onError(Throwable t) {
                    ManagedChannel llmanagedChannel = managedChannel;
                    if (llmanagedChannel != null) {
                        try {
                            llmanagedChannel.shutdownNow();
                        } catch (Exception e) {
                            logger.warn(format("Received error on %s, attempts %d. %s shut down of grpc channel.",
                                    PeerEventServiceClient.this.toString(), peer == null ? -1 : peer.getReconnectCount(), e.getMessage()), e);
                        }
                        managedChannel = null;
                    }
                    if (!shutdown) {
                        final long reconnectCount = peer.getReconnectCount();
                        if (PEER_EVENT_RECONNECTION_WARNING_RATE > 1 && reconnectCount % PEER_EVENT_RECONNECTION_WARNING_RATE == 1) {
                            logger.warn(format("Received error on  %s, attempts %d. %s",
                                    PeerEventServiceClient.this.toString(), reconnectCount, t.getMessage()));

                        } else {
                            logger.trace(format("Received error on %s, attempts %d. %s",
                                    PeerEventServiceClient.this.toString(), reconnectCount, t.getMessage()));

                        }

                        if (retry.getAndSet(false)) {
                            peer.reconnectPeerEventServiceClient(PeerEventServiceClient.this, t);
                        }

                    }
                    finishLatch.countDown();
                }

                @Override
                public void onCompleted() {
                    logger.debug(format("DeliverResponse onCompleted %s setting done.",
                            PeerEventServiceClient.this.toString()));
                    //            done = true;
                    //There should have been a done before this...
                    finishLatch.countDown();
                }
            };

            nso = filterBlock ? broadcast.deliverFiltered(so) : broadcast.deliver(so);

            nso.onNext(envelope);

            // try {
            if (!finishLatch.await(peerEventRegistrationWaitTimeMilliSecs, TimeUnit.MILLISECONDS)) {
                PeerEventingServiceException ex = new PeerEventingServiceException(format(
                        "Channel %s connect time exceeded for peer eventing service %s, timed out at %d ms.", channelName, name, peerEventRegistrationWaitTimeMilliSecs));
                ex.setTimedOut(peerEventRegistrationWaitTimeMilliSecs);
                logger.warn(toString() + " " + ex.getMessage());
                throwableList.add(0, ex);

            }
            logger.trace(toString() + " done waiting for reply!");

            if (!throwableList.isEmpty()) {
                ManagedChannel llmanagedChannel = managedChannel;
                if (llmanagedChannel != null) {
                    llmanagedChannel.shutdownNow();
                    managedChannel = null;
                }
                Throwable throwable = throwableList.get(0);
                if (retry.getAndSet(false)) {
                    peer.reconnectPeerEventServiceClient(this, throwable);
                }

            }

        } catch (InterruptedException e) {
            ManagedChannel llmanagedChannel = managedChannel;
            if (llmanagedChannel != null) {
                llmanagedChannel.shutdownNow();
                managedChannel = null;
            }
            logger.error(toString() + " error message: " + e.getMessage(), e); // not likely

            if (retry.getAndSet(false)) {
                peer.reconnectPeerEventServiceClient(this, e);
            }

        } finally {
            if (null != nso) {

                try {
                    nso.onCompleted();
                } catch (Exception e) {  //Best effort only report on debug
                    logger.debug(format("Exception completing connect with %s %s",
                            toString(), e.getMessage()), e);
                }

            }
        }
    }
874887
@Override
    public java.util.concurrent.Future<SendMessageBatchResult> sendMessageBatchAsync(String queueUrl, java.util.List<SendMessageBatchRequestEntry> entries,
            com.amazonaws.handlers.AsyncHandler<SendMessageBatchRequest, SendMessageBatchResult> asyncHandler) {

        return sendMessageBatchAsync(new SendMessageBatchRequest().withQueueUrl(queueUrl).withEntries(entries), asyncHandler);
    }
543727
protected Supplier<CompletableFuture<ExecutionResult>> supplyAsync(
      Supplier<CompletableFuture<ExecutionResult>> supplier, Scheduler scheduler, FailsafeFuture<Object> future) {
    return () -> {
      ExecutionResult result = preExecute();
      if (result != null)
        return CompletableFuture.completedFuture(result);

      return supplier.get().thenCompose(s -> postExecuteAsync(s, scheduler, future));
    };
  }
1251027
public static function deliver($clazz, $method = "run", $params = null, $task_id = -1) {
        $app = Ioc::get(Application::class);
        /* @var $deliver TaskConfig */
        $deliver = Ioc::get(TaskConfig::class);
        $bean = ['clazz' => $clazz,
                 'method' => $method,
                 'params' => $params,
                 'config' => $deliver->getTaskInitConfig()];
        if (IS_SWOOLE) {
            $app->server->task($bean, $task_id);
        }
    }