spring boot 中autobahn-java(wamp)的使用

        最近在后端开发中遇到了须要和前端进行通讯的问题。由于之前用到过crossbar的 wamp (此处的wamp可不是apache+Mysql/MariaDB+Perl/PHP/Python,在搜索wamp时会有混淆，建议搜索 wamp protocol)， 可是国内用这个东西的真的是太少了，最多就是用在android上进行消息推送（又或许本人看的太少）。 网上关于spring 调用crossbar 提供的官方包 antobahn-java的资料能够说没有（我反正是没搜到过）。因此在完成功能后，特地将调用方式贴出来，以备本身使用和方便其余有须要的猿们:sunglasses:。 crossbar官方网站

wamp简单介绍

        不说废话，wamp在消息转发这方面使用 发布/订阅 模式 即 publish/subscribe。 在wamp中有 topic的概念。发布者，订阅者订阅相同主题才可达到消息转发的功能。关于服务器的搭建就不作叙述，跟着官网很容易搭建出来。 wamp有域的概念 即下文提到的realm，发布订阅双发需在同一realm中才可正常通讯。

干货

        在简单介绍里，咱们提到了 wamp使用 发布/订阅模式来进行消息转发，那么咱们接下用spring boot来实现官方包 autobahn-java的调用。 maven地址

<dependency>
            <groupId>io.crossbar.autobahn</groupId>
            <artifactId>autobahn-java</artifactId>
            <version>18.5.1</version>
        </dependency>
复制代码

wamp配置

代码以下：

package com.lichaobao.springwamp.config;

import io.crossbar.autobahn.wamp.Client;
import io.crossbar.autobahn.wamp.Session;
import io.crossbar.autobahn.wamp.interfaces.ISession;
import io.crossbar.autobahn.wamp.types.CloseDetails;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/** * @author lichaobao * @date 2019/1/14 * @QQ 1527563274 */
@Configuration
public class WAMPConfig {
    @Value("${wamp.url}")
    private String url;
    @Value("${wamp.realm}")
    private String realm;
    private static volatile boolean canSend = false;
    @Bean
    public Session getSession(){
        Session session = new Session();
        session.addOnConnectListener(new ISession.OnConnectListener() {
            @Override
            public void onConnect(Session session) {
                int a = 0;
                if(session.isConnected()){
                    canSend = true;//链接后 检查是否能发送消息 并把canSend置为true
                }
                System.out.println("connect");
            }
        });
        session.addOnLeaveListener(new ISession.OnLeaveListener() {
            @Override
            public void onLeave(Session session, CloseDetails closeDetails) {
                System.out.println("onLeave");
                canSend = false;
            }
        });
        session.addOnDisconnectListener(new ISession.OnDisconnectListener() {
            @Override
            public void onDisconnect(Session session, boolean b) {
                System.out.println("disconnect");
            }
        });
        Client client = new Client(session,url,realm);
        client.connect();
        return session;
    }

    /** * 检查是否能发送消息 * @return */
    public static boolean isCanSend() {
        return canSend;
    }
}

复制代码

工具示例

package com.wugeek.deepassserver.utils;

import com.wugeek.deepassserver.config.WAMPConfig;
import io.crossbar.autobahn.wamp.Session;
import io.crossbar.autobahn.wamp.interfaces.ISession;
import io.crossbar.autobahn.wamp.types.CloseDetails;
import io.crossbar.autobahn.wamp.types.Publication;
import io.crossbar.autobahn.wamp.types.SessionDetails;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;

/** * @author lichaobao * @date 2019/1/14 * @QQ 1527563274 */
@Component
public class WAMPUtils {
    @Autowired
    Session session;
    public String sendMsg(String topic,String content){
        if(WAMPConfig.isCanSend()){
          CompletableFuture<Publication> pubFuture =  session.publish(topic,content);
          pubFuture.thenAccept(accept->{
              System.out.println("publish to "+topic);
          });
          return "ok";
        }else{
            return "error";
        }
    }
}
public String subscribe(String topic){
        if(WAMPConfig.isCanSend()){
            CompletableFuture<Subscription> subFuture = session.subscribe(topic,
                    this::onEvent);
            subFuture.whenComplete((subscription, throwable) -> {
                if (throwable == null) {
                    // We have successfully subscribed.
                    System.out.println("Subscribed to topic " + subscription.topic);
                } else {
                    // Something went bad.
                    throwable.printStackTrace();
                }
            });
            return "ok";
        }else{
            return "error";
        }
    }
    private void onEvent(List<Object> args) {
        System.out.println(String.format("Got event: %s", args.get(0)));
    }

复制代码

-完-