src/main/java/com/io7m/aeron_guide/take4/aeron_messaging/server/AeronMessagingServer.java
/**
     * The main() f-n is a simple example on how to instantiate and run the
     * AeronMessagingServer class in its own thread. It is not an "entry point"
     * of the class and is here only for demo purposes. The actual "entry point"
     * is the run() method, which will be engaged when you call thread.start()
     * (see main() for details).
     *
     * @param args Command-line arguments
     *
     * @throws Exception On any error
     */
The main() f-n is a simple example on how to instantiate and run the
AeronMessagingServer class in its own thread. It is not an "entry point"
of the class and is here only for demo purposes. The actual "entry point"
is the run() method, which will be engaged when you call thread.start()
for details).

@param args Command-line arguments

@throws Exception On any error
The main () f - n is a simple example on how to instantiate and run the AeronMessagingServer class in its own thread . It is not an " entry point " of the class and is here only for demo purposes . The actual " entry point " is the run () method which will be engaged when you call thread . start () for details ) . @param args Command - line arguments @throws Exception On any error
public static void main(
            final String[] args)
            throws Exception {
        if (args.length < 6) {
            LOG.error(
                    "usage: directory local-address local-initial-data-port local-initial-control-port local-clients-base-port client-count");
            System.exit(1);
        }
        final Path directory = Paths.get(args[0]);
        final InetAddress local_address = InetAddress.getByName(args[1]);
        final int local_initial_data_port = Integer.parseUnsignedInt(args[2]);
        final int local_initial_control_port = Integer.parseUnsignedInt(args[3]);
        final int local_clients_base_port = Integer.parseUnsignedInt(args[4]);
        final int client_count = Integer.parseUnsignedInt(args[5]);
        final AeronMessagingServerConfiguration configuration
                = ImmutableAeronMessagingServerConfiguration.builder()
                        .baseDirectory(directory)
                        .localAddress(local_address)
                        .localInitialPort(local_initial_data_port)
                        .localInitialControlPort(local_initial_control_port)
                        .localClientsBasePort(local_clients_base_port)
                        .clientMaximumCount(client_count)
                        .maximumConnectionsPerAddress(3)
                        .build();
        final AeronMessagingServer aeron_messaging_server = create(Clock.systemUTC(), configuration);
        aeron_messaging_server.aeron_messaging_server_thread = new Thread(aeron_messaging_server);
        aeron_messaging_server.aeron_messaging_server_thread.start();
         Main consumer loop (it uses aeron_messaging_server to send receive messages from the client(s))
        LOG.debug("Main thread keep chaga away while AeronMessagingClient is working in it's own separate thread...");
         Some activity on the main thread, which shovels/sends messages from/to aeron_messaging_client
        long total_incoming_public_messages_count = 0;
        long total_incoming_private_messages_count = 0;
        long iteration_incoming_public_messages_count = 0;
        long iteration_incoming_private_messages_count = 0;
        Clock clock = Clock.systemUTC();
        long last_stats_sent_epoch_ms = 0;  keep track of exact ms when report was generated to avoid sending it multiple times (loop is fast)
        Instant main_loop_start_instant = clock.instant();
        long main_loop_start_epoch_ms = main_loop_start_instant.toEpochMilli();
        long main_loop_iterations_count = 0;
         Sometimes we can't read just 1 incoming message in 1 main loop iteration since
         it might fundamentally limit the throughput of our application (for example the main
         loop of our app does lots of other things, so processing only one message per one
         iteration can be too cpu-expensive).
         On the other hand we can't process ALL incoming messages, since the queue might
         be already "too full" and a new messages stream is constantly coming thus we would be blocked
         at this attempt to process all the imcoming messages at once.
         As a solution let's introduce new variable: "number_of_messages_to_read_in_one_go_limit",
         which would allow us to limit the max number of messages to process in one main loop iteration.
         Reasonable values would be a couple hundreds (unless your messages are some huge JSON objects,
         in which case you probably want to make this value down to a couple dosen). Play with it.
        int number_of_messages_to_read_in_one_go_limit = 100;
         This "main loop" mimicking the demo application (aka "consumer loop"), that created AeronMessagingServer instance.
        while (true) {
            main_loop_iterations_count++;
             Get current iteration time
            Instant now_instant = clock.instant();
            long now_epoch_ms = now_instant.toEpochMilli();
             Per-iteration counts
            iteration_incoming_public_messages_count = 0;
            iteration_incoming_private_messages_count = 0;
             Try to get public-channel (all-clients) message
            String incoming_public_message = aeron_messaging_server.get_all_clients_control_channel_message();
            if (incoming_public_message != null) {
                total_incoming_public_messages_count++;
                iteration_incoming_public_messages_count++;
            }
             Try to get private (per agent channels) messages
            Enumeration<Integer> clients_session_ids = aeron_messaging_server.list_connected_clients_session_ids();
            while (clients_session_ids.hasMoreElements()) {
                 Get i-th connected client session_id
                Integer client_session_id = clients_session_ids.nextElement();
                 Count how many private messages we've got from this particular
                 session_id during current main loop iteration. We need to know
                 this count to be able to track number_of_messages_to_read_in_one_go_limit.
                long ith_session_private_message_iteration_count = 0;
                 Try to read N private messages from i-th connected client (by given session_id)
                 Keep shoveling N private messages from this session_id until
                 we read them all or until we hit the limit (number_of_messages_to_read_in_one_go_limit)
                String private_message = aeron_messaging_server.get_private_message(client_session_id);
                while (private_message != null && ith_session_private_message_iteration_count < number_of_messages_to_read_in_one_go_limit) {
                    ith_session_private_message_iteration_count++;
                     For better stats, let's consider server's main_loop_start_epoch_ms when we got 1st private message from the client.
                     TODO: redo the whole stats collection logic, so it works properly even if the server is running for few weeks and
                     the client just connected. The messages rate calculation can not rely on the server lifetime number of seconds...
                    if (total_incoming_private_messages_count == 0) {
                         First ever private message received!
                        main_loop_start_instant = clock.instant();
                        main_loop_start_epoch_ms = main_loop_start_instant.toEpochMilli();
                    }
                    total_incoming_private_messages_count++;
                    iteration_incoming_private_messages_count++;
                     Get the next message, unitll we whovel them all to zero:
                    private_message = aeron_messaging_server.get_private_message(client_session_id);
                }
            }
             Only sleep 1ms in case we haven't received any messages, otherwise it is a "busy loop"
            if (iteration_incoming_public_messages_count == 0 && iteration_incoming_private_messages_count == 0) {
                Thread.sleep(1);
                 Main thread checks periodically if messaging thread is still runnig.
                 Check if aeron_messaging_thread is still running
                 OR shall we use aeron_messaging_thread.getState() ?
                if (!aeron_messaging_server.aeron_messaging_server_thread.isAlive()) {
                    break;
                }
            }
             Log "main loop stats" once in a while
            if (now_epoch_ms > last_stats_sent_epoch_ms + 5000
                    && last_stats_sent_epoch_ms != now_epoch_ms) {
                 Time to generate some stats on the "consumer main loop"
                long main_loop_run_duration_ms = now_epoch_ms - main_loop_start_epoch_ms + 1;   +1 is a cheesy way to avoid /0 and it won't matter after few seconds run
                long average_rx_private_messages_per_s = total_incoming_private_messages_count * 1000 / main_loop_run_duration_ms;
                long main_loop_iterations_rate_per_ms = main_loop_iterations_count / main_loop_run_duration_ms;
                LOG.debug("Main consumer loop: thread: " + Thread.currentThread().toString()
                        + " total_incoming_public_messages_count: " + total_incoming_public_messages_count
                        + " total_incoming_private_messages_count: " + total_incoming_private_messages_count
                        + " main_loop_iterations_rate_per_ms: " + main_loop_iterations_rate_per_ms
                        + " average_rx_private_messages_per_s: " + average_rx_private_messages_per_s + " <--- !!!"
                );
                last_stats_sent_epoch_ms = now_epoch_ms;
            }
        }
        LOG.debug("AeronMessagingServer.main(): main loop is complete. That's all falks!");
         Need to close Aeron and MediaDriver, otherwise the program will not terminate.
        aeron_messaging_server.close();
    }
public static void main ( final String [ ] args ) throws Exception { if ( args . length < 6 ) { LOG . error ( "usage: directory local-address local-initial-data-port local-initial-control-port local-clients-base-port client-count" ) ; System . exit ( 1 ) ; } final Path directory = Paths . get ( args [ 0 ] ) ; final InetAddress local_address = InetAddress . getByName ( args [ 1 ] ) ; final int local_initial_data_port = Integer . parseUnsignedInt ( args [ 2 ] ) ; final int local_initial_control_port = Integer . parseUnsignedInt ( args [ 3 ] ) ; final int local_clients_base_port = Integer . parseUnsignedInt ( args [ 4 ] ) ; final int client_count = Integer . parseUnsignedInt ( args [ 5 ] ) ; final AeronMessagingServerConfiguration configuration = ImmutableAeronMessagingServerConfiguration . builder ( ) . baseDirectory ( directory ) . localAddress ( local_address ) . localInitialPort ( local_initial_data_port ) . localInitialControlPort ( local_initial_control_port ) . localClientsBasePort ( local_clients_base_port ) . clientMaximumCount ( client_count ) . maximumConnectionsPerAddress ( 3 ) . build ( ) ; final AeronMessagingServer aeron_messaging_server = create ( Clock . systemUTC ( ) , configuration ) ; aeron_messaging_server . aeron_messaging_server_thread = new Thread ( aeron_messaging_server ) ; aeron_messaging_server . aeron_messaging_server_thread . start ( ) ; LOG . debug ( "Main thread keep chaga away while AeronMessagingClient is working in it's own separate thread..." ) ; long total_incoming_public_messages_count = 0 ; long total_incoming_private_messages_count = 0 ; long iteration_incoming_public_messages_count = 0 ; long iteration_incoming_private_messages_count = 0 ; Clock clock = Clock . systemUTC ( ) ; long last_stats_sent_epoch_ms = 0 ; Instant main_loop_start_instant = clock . instant ( ) ; long main_loop_start_epoch_ms = main_loop_start_instant . toEpochMilli ( ) ; long main_loop_iterations_count = 0 ; int number_of_messages_to_read_in_one_go_limit = 100 ; while ( true ) { main_loop_iterations_count ++ ; Instant now_instant = clock . instant ( ) ; long now_epoch_ms = now_instant . toEpochMilli ( ) ; iteration_incoming_public_messages_count = 0 ; iteration_incoming_private_messages_count = 0 ; String incoming_public_message = aeron_messaging_server . get_all_clients_control_channel_message ( ) ; if ( incoming_public_message != null ) { total_incoming_public_messages_count ++ ; iteration_incoming_public_messages_count ++ ; } Enumeration < Integer > clients_session_ids = aeron_messaging_server . list_connected_clients_session_ids ( ) ; while ( clients_session_ids . hasMoreElements ( ) ) { Integer client_session_id = clients_session_ids . nextElement ( ) ; long ith_session_private_message_iteration_count = 0 ; String private_message = aeron_messaging_server . get_private_message ( client_session_id ) ; while ( private_message != null && ith_session_private_message_iteration_count < number_of_messages_to_read_in_one_go_limit ) { ith_session_private_message_iteration_count ++ ; if ( total_incoming_private_messages_count == 0 ) { main_loop_start_instant = clock . instant ( ) ; main_loop_start_epoch_ms = main_loop_start_instant . toEpochMilli ( ) ; } total_incoming_private_messages_count ++ ; iteration_incoming_private_messages_count ++ ; private_message = aeron_messaging_server . get_private_message ( client_session_id ) ; } } if ( iteration_incoming_public_messages_count == 0 && iteration_incoming_private_messages_count == 0 ) { Thread . sleep ( 1 ) ; if ( ! aeron_messaging_server . aeron_messaging_server_thread . isAlive ( ) ) { break ; } } if ( now_epoch_ms > last_stats_sent_epoch_ms + 5000 && last_stats_sent_epoch_ms != now_epoch_ms ) { long main_loop_run_duration_ms = now_epoch_ms - main_loop_start_epoch_ms + 1 ; long average_rx_private_messages_per_s = total_incoming_private_messages_count * 1000 / main_loop_run_duration_ms ; long main_loop_iterations_rate_per_ms = main_loop_iterations_count / main_loop_run_duration_ms ; LOG . debug ( "Main consumer loop: thread: " + Thread . currentThread ( ) . toString ( ) + " total_incoming_public_messages_count: " + total_incoming_public_messages_count + " total_incoming_private_messages_count: " + total_incoming_private_messages_count + " main_loop_iterations_rate_per_ms: " + main_loop_iterations_rate_per_ms + " average_rx_private_messages_per_s: " + average_rx_private_messages_per_s + " <--- !!!" ) ; last_stats_sent_epoch_ms = now_epoch_ms ; } } LOG . debug ( "AeronMessagingServer.main(): main loop is complete. That's all falks!" ) ; aeron_messaging_server . close ( ) ; }
The main() f-n is a simple example on how to instantiate and run the
 AeronMessagingServer class in its own thread.
The main () f - n is a simple example on how to instantiate and run the AeronMessagingServer class in its own thread .
// Parse command-line args // Create AeronMessagingServerConfiguration object // Start aeron_messaging_server in it's own thread, we'll use it's public methods to enqueue (send) / dequeue (receive) messages // // Main consumer loop (it uses aeron_messaging_server to send receive messages from the client(s)) // // Some activity on the main thread, which shovels/sends messages from/to aeron_messaging_client // keep track of exact ms when report was generated to avoid sending it multiple times (loop is fast) // Sometimes we can't read just 1 incoming message in 1 main loop iteration since // it might fundamentally limit the throughput of our application (for example the main // loop of our app does lots of other things, so processing only one message per one // iteration can be too cpu-expensive). // On the other hand we can't process ALL incoming messages, since the queue might // be already "too full" and a new messages stream is constantly coming thus we would be blocked // at this attempt to process all the imcoming messages at once. // As a solution let's introduce new variable: "number_of_messages_to_read_in_one_go_limit", // which would allow us to limit the max number of messages to process in one main loop iteration. // Reasonable values would be a couple hundreds (unless your messages are some huge JSON objects, // in which case you probably want to make this value down to a couple dosen). Play with it. // This "main loop" mimicking the demo application (aka "consumer loop"), that created AeronMessagingServer instance. // Get current iteration time // Per-iteration counts // Try to get public-channel (all-clients) message // Try to get private (per agent channels) messages // Get i-th connected client session_id // Count how many private messages we've got from this particular // session_id during current main loop iteration. We need to know // this count to be able to track number_of_messages_to_read_in_one_go_limit. // Try to read N private messages from i-th connected client (by given session_id) // Keep shoveling N private messages from this session_id until // we read them all or until we hit the limit (number_of_messages_to_read_in_one_go_limit) // For better stats, let's consider server's main_loop_start_epoch_ms when we got 1st private message from the client. // TODO: redo the whole stats collection logic, so it works properly even if the server is running for few weeks and // the client just connected. The messages rate calculation can not rely on the server lifetime number of seconds... // First ever private message received! // Get the next message, unitll we whovel them all to zero: // Only sleep 1ms in case we haven't received any messages, otherwise it is a "busy loop" // Main thread checks periodically if messaging thread is still runnig. // Check if aeron_messaging_thread is still running // OR shall we use aeron_messaging_thread.getState() ? // Log "main loop stats" once in a while // Time to generate some stats on the "consumer main loop" // +1 is a cheesy way to avoid /0 and it won't matter after few seconds run // Need to close Aeron and MediaDriver, otherwise the program will not terminate.