library/src/main/java/com/kdgregory/aws/utils/cloudwatch/CloudWatchLogsWriter.java
/**
     *  Attempts to send a batch of messages to the service. If successful, returns
     *  the sequence token for the next batch. Allows any exceptions to propagate.
     */
Attempts to send a batch of messages to the service. If successful, returns
the sequence token for the next batch. Allows any exceptions to propagate.
Attempts to send a batch of messages to the service . If successful returns the sequence token for the next batch . Allows any exceptions to propagate .
private String attemptToSend(List<QueuedMessage> batch, String uploadSequenceToken)
    {
        if (logBatches && logger.isDebugEnabled())
        {
            logger.debug("sending batch of " + batch.size() + " events to " + logGroupName + " / " + logStreamName);
        }
        List<InputLogEvent> events = new ArrayList<InputLogEvent>(batch.size());
        for (QueuedMessage message : batch)
        {
            events.add(new InputLogEvent()
                      .withTimestamp(message.timestamp)
                      .withMessage(message.message));
        }
        PutLogEventsRequest request = new PutLogEventsRequest(logGroupName, logStreamName, events)
                                      .withSequenceToken(uploadSequenceToken);
        PutLogEventsResult response = client.putLogEvents(request);
        int messagesRejected = 0;
        RejectedLogEventsInfo rejectInfo = response.getRejectedLogEventsInfo();
        if (rejectInfo != null)
        {
            int expired = rejectInfo.getExpiredLogEventEndIndex() != null
                        ? rejectInfo.getExpiredLogEventEndIndex().intValue() + 1
                        : 0;
            int tooOld  = rejectInfo.getTooOldLogEventEndIndex() != null
                        ? rejectInfo.getTooOldLogEventEndIndex().intValue() + 1
                        : 0;
            int tooNew  = rejectInfo.getTooNewLogEventStartIndex() != null
                        ? batch.size() - rejectInfo.getTooNewLogEventStartIndex().intValue()
                        : 0;
            messagesRejected = Math.max(expired, tooOld) + tooNew;
            logger.warn("rejected " + messagesRejected
                        + " messages on stream " + logGroupName + " / " + logStreamName + ": "
                        + Math.max(expired, tooOld) + " expired/too old, " + tooNew + " too new");
        }
        batchCount.incrementAndGet();
        totalMessagesSent.addAndGet(batch.size());
        totalMessagesRejected.addAndGet(messagesRejected);
        return response.getNextSequenceToken();
    }
private String attemptToSend ( List < QueuedMessage > batch , String uploadSequenceToken ) { if ( logBatches && logger . isDebugEnabled ( ) ) { logger . debug ( "sending batch of " + batch . size ( ) + " events to " + logGroupName + " / " + logStreamName ) ; } List < InputLogEvent > events = new ArrayList < InputLogEvent > ( batch . size ( ) ) ; for ( QueuedMessage message : batch ) { events . add ( new InputLogEvent ( ) . withTimestamp ( message . timestamp ) . withMessage ( message . message ) ) ; } PutLogEventsRequest request = new PutLogEventsRequest ( logGroupName , logStreamName , events ) . withSequenceToken ( uploadSequenceToken ) ; PutLogEventsResult response = client . putLogEvents ( request ) ; int messagesRejected = 0 ; RejectedLogEventsInfo rejectInfo = response . getRejectedLogEventsInfo ( ) ; if ( rejectInfo != null ) { int expired = rejectInfo . getExpiredLogEventEndIndex ( ) != null ? rejectInfo . getExpiredLogEventEndIndex ( ) . intValue ( ) + 1 : 0 ; int tooOld = rejectInfo . getTooOldLogEventEndIndex ( ) != null ? rejectInfo . getTooOldLogEventEndIndex ( ) . intValue ( ) + 1 : 0 ; int tooNew = rejectInfo . getTooNewLogEventStartIndex ( ) != null ? batch . size ( ) - rejectInfo . getTooNewLogEventStartIndex ( ) . intValue ( ) : 0 ; messagesRejected = Math . max ( expired , tooOld ) + tooNew ; logger . warn ( "rejected " + messagesRejected + " messages on stream " + logGroupName + " / " + logStreamName + ": " + Math . max ( expired , tooOld ) + " expired/too old, " + tooNew + " too new" ) ; } batchCount . incrementAndGet ( ) ; totalMessagesSent . addAndGet ( batch . size ( ) ) ; totalMessagesRejected . addAndGet ( messagesRejected ) ; return response . getNextSequenceToken ( ) ; }
Attempts to send a batch of messages to the service.
Attempts to send a batch of messages to the service .