src/main/java/com/io7m/aeron_guide/take4/aeron_messaging/common/MessagesHelper.java
/**
     * Send the given message to the given publication. If the publication fails
     * to accept the message, the method will retry {@code 5} times, waiting
     * {@code 100} milliseconds each time, before throwing an exception.
     *
     * @param publication The publication
     * @param buffer A buffer that will hold the message for sending
     * @param text The message
     *
     * @return The new publication stream position
     *
     * @throws IOException If the message cannot be sent
     */
Send the given message to the given publication. If the publication fails
to accept the message, the method will retry  5 times, waiting
100 milliseconds each time, before throwing an exception.

@param publication The publication
@param buffer A buffer that will hold the message for sending
@param text The message

@return The new publication stream position

@throws IOException If the message cannot be sent
Send the given message to the given publication . If the publication fails to accept the message the method will retry 5 times waiting 100 milliseconds each time before throwing an exception . @param publication The publication @param buffer A buffer that will hold the message for sending @param text The message @return The new publication stream position @throws IOException If the message cannot be sent
public static long send_message(
            final Publication publication,
            final UnsafeBuffer buffer,
            final String text)
            throws IOException {
        Objects.requireNonNull(publication, "publication");
        Objects.requireNonNull(buffer, "buffer");
        Objects.requireNonNull(text, "text");
        final byte[] value = text.getBytes(UTF_8);
        buffer.putBytes(0, value);
        long result_code = 0L;
        for (int index = 0; index < 5; ++index) {
            result_code = publication.offer(buffer, 0, value.length);
            if (result_code < 0L) {
                try {
                    Thread.sleep(50L);
                } catch (final InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                continue;
            }
            return result_code;
        }
        if (result_code == Publication.NOT_CONNECTED) {
            throw new PublicationNotConnected();
        }
        if (result_code == Publication.ADMIN_ACTION) {
            throw new PublicationAdminAction();
        }
        if (result_code == Publication.BACK_PRESSURED) {
            throw new PublicationBackPressured();
        }
        if (result_code == Publication.CLOSED) {
            throw new PublicationClosed();
        }
        if (result_code == Publication.MAX_POSITION_EXCEEDED) {
            throw new PublicationMaxPositionExceeded();
        }
        throw new IllegalStateException();
    }
public static long send_message ( final Publication publication , final UnsafeBuffer buffer , final String text ) throws IOException { Objects . requireNonNull ( publication , "publication" ) ; Objects . requireNonNull ( buffer , "buffer" ) ; Objects . requireNonNull ( text , "text" ) ; final byte [ ] value = text . getBytes ( UTF_8 ) ; buffer . putBytes ( 0 , value ) ; long result_code = 0L ; for ( int index = 0 ; index < 5 ; ++ index ) { result_code = publication . offer ( buffer , 0 , value . length ) ; if ( result_code < 0L ) { try { Thread . sleep ( 50L ) ; } catch ( final InterruptedException e ) { Thread . currentThread ( ) . interrupt ( ) ; } continue ; } return result_code ; } if ( result_code == Publication . NOT_CONNECTED ) { throw new PublicationNotConnected ( ) ; } if ( result_code == Publication . ADMIN_ACTION ) { throw new PublicationAdminAction ( ) ; } if ( result_code == Publication . BACK_PRESSURED ) { throw new PublicationBackPressured ( ) ; } if ( result_code == Publication . CLOSED ) { throw new PublicationClosed ( ) ; } if ( result_code == Publication . MAX_POSITION_EXCEEDED ) { throw new PublicationMaxPositionExceeded ( ) ; } throw new IllegalStateException ( ) ; }
Send the given message to the given publication.
Send the given message to the given publication .
//    LOG.trace("[{}] send: {}", Integer.toString(pub.sessionId()), text); // Note: when we shut down server, and the client is trying to "publication.offer()" then // 1st 2 results will be result_code=-2 (as if Publication.BACK_PRESSURED) // and only attepmt 3-4-5 will have result_code=-1 (as Publication.NOT_CONNECTED). So if // you give only 2 attempts here it won't work correctly (you won't see NOT_CONNECTED and your program // will keep re-enqueueing the message as if it was BACK_PRESSUR'ed, which is wrong). Tested this on Aeron ver. 1.8.2 and ver. 1.25.1 // *** Update: *** actually it has nothing to do with number of tries, but rather timing. It takes Aeron to realize that // the connection to server is lost somewhere around: // 23:00:33.146 - started getting result_code=-2 (as if backbressured) // 23:00:36.351 - got 1st (well, and last, since exit on it) result_code=-1, which was 3 seconds later. // So it took client 3 seconds to realise -1 (NOT_CONNECTED) // If failed to send the message, let's sleep a few milliseconds and try again. //        // When connection to the server is lost the client will throw an exception: //        // ClientIOException: java.io.IOException: Could not send message: Error code: Not connected //        throw new IOException( //                "Could not send message: Error code: " + error_code_to_string(result_code)); // Let's throw different exceptions depending on the error code we got from publication.offer()