Статьи

Простой обмен сообщениями с STOMP через WebSockets с использованием Apollo

Август 22, 2018

В моем  предыдущем посте  я рассмотрел пару интересных случаев использования реализации   сообщений STOMP через  Websockect  с использованием известных брокеров сообщений,  HornetQ  и  ActiveMQ . Но тот, который я не рассмотрел, — это  Apollo,  поскольку, по моему мнению, его API является многословным и недостаточно выразительным, как для разработчика на Java. Тем не менее, чем больше времени я проводил, играя с  Аполлоном , тем больше убеждался в том, что у меня есть огромный потенциал. Так что этот пост все об  Аполлоне .

Проблема, которую мы пытаемся решить, остается той же: простое решение публикации / подписки, где веб-клиент JavaScript отправляет сообщения и прослушивает определенную тему. При получении любого сообщения клиент отображает окно с предупреждением (обратите внимание, что нам нужно использовать современный браузер с поддержкой  веб-сокетов , например,  Google Chrome  или  Mozilla Firefox ).

Давайте запачкаем руки, начав с  index.html  (который импортирует потрясающую   библиотеку JavaScript stomp.js ):

<script src="stomp.js"></script>

<script type="text/javascript">
    var client = Stomp.client( "ws://localhost:61614/stomp", "v11.stomp" );

    client.connect( "", "",
        function() {
            client.subscribe("/topic/test",
                function( message ) {
                    alert( message );
                }, 
                { priority: 9 } 
            );
   
            client.send("/topic/test", { priority: 9 }, "Pub/Sub over STOMP!");
        }
    ); 
</script>

Клиентская часть ничем не отличается, кроме названия темы, которое теперь  / topic / test . Однако сторона сервера сильно отличается. Аполлон  написан на  Scala  и охватывает асинхронную неблокирующую модель программирования. Я думаю, это очень хорошая вещь. Хотя это приносит новую парадигму для программирования, и это также не обязательно плохо. Класс  AppConfig  предназначен для настройки встроенного   брокера Apollo :

package com.example.messaging;

import java.io.File;

import org.apache.activemq.apollo.broker.Broker;
import org.apache.activemq.apollo.broker.jmx.dto.JmxDTO;
import org.apache.activemq.apollo.dto.AcceptingConnectorDTO;
import org.apache.activemq.apollo.dto.BrokerDTO;
import org.apache.activemq.apollo.dto.TopicDTO;
import org.apache.activemq.apollo.dto.VirtualHostDTO;
import org.apache.activemq.apollo.dto.WebAdminDTO;
import org.apache.activemq.apollo.stomp.dto.StompDTO;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class AppConfig {
    @Bean
    public Broker broker() throws Exception {
        final Broker broker = new Broker();
    
        // Configure STOMP over WebSockects connector
        final AcceptingConnectorDTO ws = new AcceptingConnectorDTO();
        ws.id = "ws";
        ws.bind = "ws://localhost:61614";  
        ws.protocols.add( new StompDTO() );

        // Create a topic with name 'test'
        final TopicDTO topic = new TopicDTO();
        topic.id = "test";
  
        // Create virtual host (based on localhost)
        final VirtualHostDTO host = new VirtualHostDTO();
        host.id = "localhost";  
        host.topics.add( topic );
        host.host_names.add( "localhost" );
        host.host_names.add( "127.0.0.1" );
        host.auto_create_destinations = false;
  
        // Create a web admin UI (REST) accessible at: http://localhost:61680/api/index.html#!/ 
        final WebAdminDTO webadmin = new WebAdminDTO();
        webadmin.bind = "http://localhost:61680";

        // Create JMX instrumentation 
        final JmxDTO jmxService = new JmxDTO();
        jmxService.enabled = true;
  
        // Finally, glue all together inside broker configuration
        final BrokerDTO config = new BrokerDTO();
        config.connectors.add( ws );
        config.virtual_hosts.add( host );
        config.web_admins.add( webadmin );
        config.services.add( jmxService );
  
        broker.setConfig( config );
        broker.setTmp( new File( System.getProperty( "java.io.tmpdir" ) ) );
  
        broker.start( new Runnable() {   
            @Override
            public void run() {  
                System.out.println("The broker has been started started.");
            }
        } );
  
        return broker;
    }
}

I guess it becomes clear what I meant by verbose and not expressive enough but at least it’s easy to follow. Firstly, we are creating Websockects connector at ws://localhost:61614 and asking it to support the STOMP protocol. Then we are creating a simple topic with name test (which we refer as /topic/test on client side). Next important step is to create a virtual host and to bind topics (and queues if any) to it. The host names list is very important as the destination resolution logic heavily relies on it. In the following step we are configuring web admin UI and JMX instrumentation which provides us with access to configuration, statistics and monitoring. To check it out, please open this URL in your web browser once Apollo broker is started. And finally, by applying configuration and starting the broker we are good to go! As you can see, asynchronous programming model leads to callbacks and anonymous functions (where are you, Java 8?).

Now, when configuration is done, it’s time to look at start-up logic placed into Starter class (again, callbacks and anonymous functions are used to perform graceful shutdown logic):

package com.example.messaging;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.apache.activemq.apollo.broker.Broker;
import org.springframework.context.annotation.ConfigurableApplicationContext;

public class Starter  {
    public static void main( String[] args ) throws Exception {
        try( ConfigurableApplicationContext context = new AnnotationConfigApplicationContext( AppConfig.class ) ) {
            final Broker broker = context.getBean( Broker.class );  
            System.out.println( "Press any key to terminate ..." );
            System.in.read();    
         
            final CountDownLatch latch = new CountDownLatch( 1 );
            broker.stop( new Runnable() { 
                @Override
                public void run() {  
                    System.out.println("The broker has been stopped.");
                    latch.countDown();
                }
            } );
         
            // Gracefully stop the broker
            if( !latch.await( 1, TimeUnit.SECONDS ) ) {
                System.out.println("The broker hasn't been stopped, exiting anyway ...");
            }
        }
    }
}

As with the previous examples, after running Starter class and opening index.html in the browser, we should see something like that:

Great, it works just fine! I am pretty sure that just rewriting the code in Scala, this Apollo API usage example will look much more compact and concise. In any case, I think Apollo message broker is definitely worth to consider if you are looking for prominent messaging architecture.

All sources are available on GitHub: Apollo example.

38
Share: