在服务器端使用 Spring 框架在 android 中设置 Stomp 客户端

Set up a Stomp client in android with Spring framework in server side

提问by EPerrin95

I am developing an android application that exchanges data with a jetty server configured in Spring. To obtain a more dynamic android application, i am trying to use WebSocket protocol with Stomp messages.

我正在开发一个 android 应用程序，它与在 Spring 中配置的码头服务器交换数据。为了获得更动态的 android 应用程序，我尝试将 WebSocket 协议与 Stomp 消息一起使用。

In order to realize this stuff, i config a web socket message broker in spring :

为了实现这些东西，我在 spring 中配置了一个 web socket 消息代理：

@Configuration
//@EnableScheduling
@ComponentScan(
        basePackages="project.web",
        excludeFilters = @ComponentScan.Filter(type= FilterType.ANNOTATION, value = Configuration.class)
)
@EnableWebSocketMessageBroker
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        config.enableSimpleBroker("/message");
        config.setApplicationDestinationPrefixes("/app");
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/client");
    }
}

and a SimpMessageSendingOperationsin Spring controller to send message from server to client :

和一个SimpMessageSendingOperations在 Spring 控制器中从服务器向客户端发送消息：

@Controller
public class MessageAddController {
    private final Log log = LogFactory.getLog(MessageAddController.class);

    private SimpMessageSendingOperations messagingTemplate;

    private UserManager userManager;

    private MessageManager messageManager;

    @Autowired
    public MessageAddController(SimpMessageSendingOperations messagingTemplate, 
            UserManager userManager, MessageManager messageManager){
        this.messagingTemplate = messagingTemplate;
        this.userManager = userManager;
        this.messageManager = messageManager;
    }

    @RequestMapping("/Message/Add")
    @ResponseBody
    public SimpleMessage addFriendship(
            @RequestParam String content,
            @RequestParam Long otherUser_id
    ){
        if(log.isInfoEnabled())
            log.info("Execute MessageAdd action");
        SimpleMessage simpleMessage;

        try{
            User curentUser = userManager.getCurrentUser();
            User otherUser = userManager.findUser(otherUser_id);

            Message message = new Message();
            message.setContent(content);
            message.setUserSender(curentUser);
            message.setUserReceiver(otherUser);

            messageManager.createMessage(message);          
            Message newMessage = messageManager.findLastMessageCreated();

            messagingTemplate.convertAndSend( 
                    "/message/add", newMessage);//send message through websocket

            simpleMessage = new SimpleMessage(null, newMessage);
        } catch (Exception e) {
            if(log.isErrorEnabled())
                log.error("A problem of type : " + e.getClass() 
                        + " has occured, with message : " + e.getMessage());
            simpleMessage = new SimpleMessage(
                            new SimpleException(e.getClass(), e.getMessage()), null);
        }
        return simpleMessage;
    }
}

When i test this configuration in a web browser with stomp.js, I haven't any problem : messages are perfectly exchanged between web browser and Jetty server. The JavaScript code using for web browser test :

当我使用 stomp.js 在 web 浏览器中测试这个配置时，我没有任何问题：消息在 web 浏览器和 Jetty 服务器之间完美交换。用于 Web 浏览器测试的 JavaScript 代码：

    var stompClient = null;

    function setConnected(connected) {
        document.getElementById('connect').disabled = connected;
        document.getElementById('disconnect').disabled = !connected;
        document.getElementById('conversationDiv').style.visibility = connected ? 'visible' : 'hidden';
        document.getElementById('response').innerHTML = '';
    }

    function connect() {
        stompClient = Stomp.client("ws://YOUR_IP/client");         
        stompClient.connect({}, function(frame) {
            setConnected(true);
            stompClient.subscribe('/message/add', function(message){
                showMessage(JSON.parse(message.body).content);
            });
        });
    }

    function disconnect() {
        stompClient.disconnect();
        setConnected(false);
        console.log("Disconnected");
    }


    function showMessage(message) {
        var response = document.getElementById('response');
        var p = document.createElement('p');
        p.style.wordWrap = 'break-word';
        p.appendChild(document.createTextNode(message));
        response.appendChild(p);
    }

Problems occur when i try to use stomp in Android with libraries like gozirra, activemq-stomp or others : most of time, connection with server doesn't work. My app stop to run and, after a few minutes, i have the following message in logcat : java.net.UnknownHostException: Unable to resolve host "ws://192.168.1.39/client": No address associated with hostname, and i don't understand why. Code using Gozzira library which manages the stomp appeal in my android activity :

当我尝试在 Android 中使用带有 gozirra、activemq-stomp 或其他库的 stomp 时出现问题：大多数情况下，与服务器的连接不起作用。我的应用程序停止运行，几分钟后，我在 logcat 中有以下消息：java.net.UnknownHostException: Unable to resolve host "ws://192.168.1.39/client": No address associated with hostname，我不明白为什么。使用 Gozzira 库管理我的 android 活动中的跺脚上诉的代码：

private void stomp_test() {
    String ip = "ws://192.172.6.39/client";
    int port = 8080;

    String channel = "/message/add";
    Client c;

    try {
        c = new Client( ip, port, "", "" );
        Log.i("Stomp", "Connection established");
        c.subscribe( channel, new Listener() {
            public void message( Map header, String message ) {
                Log.i("Stomp", "Message received!!!");
              }
        });

    } catch (IOException ex) {
        Log.e("Stomp", ex.getMessage());
        ex.printStackTrace();

    } catch (LoginException ex) {
        Log.e("Stomp", ex.getMessage());
        ex.printStackTrace();
    } catch (Exception ex) {
        Log.e("Stomp", ex.getMessage());
        ex.printStackTrace();
    }

}

After some research, i found that most persons who want to use stomp over websocket with Java Client use ActiveMQ server, like in this site. But spring tools are very simple to use and it will be cool if i could keep my server layer as is it now. Someone would know how to use stomp java (Android) in client side with Spring configuration in server side?

经过一些研究，我发现大多数想要在带有 Java 客户端的 websocket 上使用 stomp 的人都使用 ActiveMQ 服务器，就像在这个站点中一样。但是 spring 工具使用起来非常简单，如果我能保持我的服务器层现在的样子会很酷。有人会知道如何在客户端使用 stomp java (Android) 和服务器端的 Spring 配置？

采纳答案by EPerrin95

I achieve to use stomp over web socket with Android and spring server.

我实现了在 Android 和 spring 服务器上使用 stomp over web socket。

To do such a thing, i used a web socket library : werbench (follow this linkto download it). To install, I used the maven command mvn installand i got back the jar in my local repository. Then, I need to add a stomp layer on the basic web socket one, but i couldn't find any stomp library in java which could manage stomp over web socket (I had to give up gozzira). So I create my own one (with stomp.js like model). Don't hesitate to ask me if you want take a look at it, but i realized it very quickly so it can not manage as much as stomp.js. Then, i need to realize an authentication with my spring server. To achieve it, i followed the indication of this site. when i get back the JSESSIONID cookie, I had just need to declare an header with this cookie in the instantiation of a werbench web socket in my stomp "library".

为了做这样的事情，我使用了一个网络套接字库：werbench（按照这个链接下载它）。为了安装，我使用了 maven 命令，mvn install并在我的本地存储库中取回了 jar。然后，我需要在基本的 web socket 上添加一个 stomp 层，但是我在 java 中找不到任何可以通过 web socket 管理 stomp 的 stomp 库（我不得不放弃 gozzira）。所以我创建了自己的一个（使用 stomp.js 之类的模型）。不要犹豫，问我是否想看一看它，但我很快意识到它，所以它无法像 stomp.js 那样管理。然后，我需要通过我的 spring 服务器实现身份验证。为了实现它，我遵循了本网站的指示. 当我取回 JSESSIONID cookie 时，我只需要在我的跺脚“库”中的 werbench Web 套接字实例化中声明一个带有此 cookie 的标头。

EDIT : this is the main class in this library, the one which manage the stomp over web socket connection :

编辑：这是这个库中的主类，它管理通过 web 套接字连接的跺脚：

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import android.util.Log;
import de.roderick.weberknecht.WebSocket;
import de.roderick.weberknecht.WebSocketEventHandler;
import de.roderick.weberknecht.WebSocketMessage;

public class Stomp {

    private static final String TAG = Stomp.class.getSimpleName();

    public static final int CONNECTED = 1;//Connection completely established
    public static final int NOT_AGAIN_CONNECTED = 2;//Connection process is ongoing
    public static final int DECONNECTED_FROM_OTHER = 3;//Error, no more internet connection, etc.
    public static final int DECONNECTED_FROM_APP = 4;//application explicitely ask for shut down the connection 

    private static final String PREFIX_ID_SUBSCIPTION = "sub-";
    private static final String ACCEPT_VERSION_NAME = "accept-version";
    private static final String ACCEPT_VERSION = "1.1,1.0";
    private static final String COMMAND_CONNECT = "CONNECT";
    private static final String COMMAND_CONNECTED = "CONNECTED";
    private static final String COMMAND_MESSAGE = "MESSAGE";
    private static final String COMMAND_RECEIPT = "RECEIPT";
    private static final String COMMAND_ERROR = "ERROR";
    private static final String COMMAND_DISCONNECT = "DISCONNECT";
    private static final String COMMAND_SEND = "SEND";
    private static final String COMMAND_SUBSCRIBE = "SUBSCRIBE";
    private static final String COMMAND_UNSUBSCRIBE = "UNSUBSCRIBE";
    private static final String SUBSCRIPTION_ID = "id";
    private static final String SUBSCRIPTION_DESTINATION = "destination";
    private static final String SUBSCRIPTION_SUBSCRIPTION = "subscription";


    private static final Set<String> VERSIONS = new HashSet<String>();
    static {
        VERSIONS.add("V1.0");
        VERSIONS.add("V1.1");
        VERSIONS.add("V1.2");
    }

    private WebSocket websocket;

    private int counter;

    private int connection;

    private Map<String, String> headers;

    private int maxWebSocketFrameSize;

    private Map<String, Subscription> subscriptions;

    private ListenerWSNetwork networkListener;

    /**
     * Constructor of a stomp object. Only url used to set up a connection with a server can be instantiate
     * 
     * @param url
     *      the url of the server to connect with
     */
    public Stomp(String url, Map<String,String> headersSetup, ListenerWSNetwork stompStates){       
        try {
            this.websocket = new WebSocket(new URI(url), null, headersSetup);
            this.counter = 0;

            this.headers = new HashMap<String, String>();
            this.maxWebSocketFrameSize = 16 * 1024;
            this.connection = NOT_AGAIN_CONNECTED;
            this.networkListener = stompStates;
            this.networkListener.onState(NOT_AGAIN_CONNECTED);
            this.subscriptions = new HashMap<String, Subscription>();

            this.websocket.setEventHandler(new WebSocketEventHandler() {        
                @Override
                public void onOpen(){
                    if(Stomp.this.headers != null){                                         
                        Stomp.this.headers.put(ACCEPT_VERSION_NAME, ACCEPT_VERSION);

                        transmit(COMMAND_CONNECT, Stomp.this.headers, null);

                        Log.d(TAG, "...Web Socket Openned");
                    }
                }

                @Override
                public void onMessage(WebSocketMessage message) {
                    Log.d(TAG, "<<< " + message.getText());
                    Frame frame = Frame.fromString(message.getText());
                    boolean isMessageConnected = false;

                    if(frame.getCommand().equals(COMMAND_CONNECTED)){
                        Stomp.this.connection = CONNECTED;
                        Stomp.this.networkListener.onState(CONNECTED);

                        Log.d(TAG, "connected to server : " + frame.getHeaders().get("server"));
                        isMessageConnected = true;

                    } else if(frame.getCommand().equals(COMMAND_MESSAGE)){
                        String subscription = frame.getHeaders().get(SUBSCRIPTION_SUBSCRIPTION);
                        ListenerSubscription onReceive = Stomp.this.subscriptions.get(subscription).getCallback();

                        if(onReceive != null){
                            onReceive.onMessage(frame.getHeaders(), frame.getBody());
                        } else{
                            Log.e(TAG, "Error : Subscription with id = " + subscription + " had not been subscribed");
                            //ACTION TO DETERMINE TO MANAGE SUBCRIPTION ERROR
                        }

                    } else if(frame.getCommand().equals(COMMAND_RECEIPT)){
                        //I DON'T KNOW WHAT A RECEIPT STOMP MESSAGE IS

                    } else if(frame.getCommand().equals(COMMAND_ERROR)){
                        Log.e(TAG, "Error : Headers = " + frame.getHeaders() + ", Body = " + frame.getBody());
                        //ACTION TO DETERMINE TO MANAGE ERROR MESSAGE

                    } else {

                    }

                    if(isMessageConnected)
                        Stomp.this.subscribe();
                }

                @Override
                public void onClose(){
                    if(connection == DECONNECTED_FROM_APP){
                        Log.d(TAG, "Web Socket disconnected");
                        disconnectFromApp();
                    } else{
                        Log.w(TAG, "Problem : Web Socket disconnected whereas Stomp disconnect method has never "
                                + "been called.");
                        disconnectFromServer();
                    }
                }

                @Override
                public void onPing() {

                }

                @Override
                public void onPong() {

                }

                @Override
                public void onError(IOException e) {
                    Log.e(TAG, "Error : " + e.getMessage());                
                }
            });
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
    }

    /**
     * Send a message to server thanks to websocket
     * 
     * @param command
     *      one of a frame property, see {@link Frame} for more details
     * @param headers
     *      one of a frame property, see {@link Frame} for more details
     * @param body
     *      one of a frame property, see {@link Frame} for more details
     */
    private void transmit(String command, Map<String, String> headers, String body){
        String out = Frame.marshall(command, headers, body);
        Log.d(TAG, ">>> " + out);
        while (true) {
            if (out.length() > this.maxWebSocketFrameSize) {
                this.websocket.send(out.substring(0, this.maxWebSocketFrameSize));
                out = out.substring(this.maxWebSocketFrameSize);
            } else {
                this.websocket.send(out);
                break;
            }
        }
    }

    /**
     * Set up a web socket connection with a server
     */
    public void connect(){
        if(this.connection != CONNECTED){
            Log.d(TAG, "Opening Web Socket...");
            try{
                this.websocket.connect();
            } catch (Exception e){
                Log.w(TAG, "Impossible to establish a connection : " + e.getClass() + ":" + e.getMessage());
            }
        }
    }

    /**
     * disconnection come from the server, without any intervention of client side. Operations order is very important
     */
    private void disconnectFromServer(){
        if(this.connection == CONNECTED){
            this.connection = DECONNECTED_FROM_OTHER;
            this.websocket.close();
            this.networkListener.onState(this.connection);
        }
    }

    /**
     * disconnection come from the app, because the public method disconnect was called
     */
    private void disconnectFromApp(){
        if(this.connection == DECONNECTED_FROM_APP){
            this.websocket.close();
            this.networkListener.onState(this.connection);
        }
    }

    /**
     * Close the web socket connection with the server. Operations order is very important
     */
    public void disconnect(){
        if(this.connection == CONNECTED){
            this.connection = DECONNECTED_FROM_APP;
            transmit(COMMAND_DISCONNECT, null, null);
        }
    }

    /**
     * Send a simple message to the server thanks to the body parameter
     * 
     * 
     * @param destination
     *      The destination through a Stomp message will be send to the server
     * @param headers
     *      headers of the message
     * @param body
     *      body of a message
     */
    public void send(String destination, Map<String,String> headers, String body){
        if(this.connection == CONNECTED){
            if(headers == null)
                headers = new HashMap<String, String>();

            if(body == null)
                body = "";

            headers.put(SUBSCRIPTION_DESTINATION, destination);

            transmit(COMMAND_SEND, headers, body);
        }
    }

    /**
     * Allow a client to send a subscription message to the server independently of the initialization of the web socket.
     * If connection have not been already done, just save the subscription
     * 
     * @param subscription
     *      a subscription object
     */
    public void subscribe(Subscription subscription){
        subscription.setId(PREFIX_ID_SUBSCIPTION + this.counter++);
        this.subscriptions.put(subscription.getId(), subscription);

        if(this.connection == CONNECTED){   
            Map<String, String> headers = new HashMap<String, String>();            
            headers.put(SUBSCRIPTION_ID, subscription.getId());
            headers.put(SUBSCRIPTION_DESTINATION, subscription.getDestination());

            subscribe(headers);
        }
    }

    /**
     * Subscribe to a Stomp channel, through messages will be send and received. A message send from a determine channel
     * can not be receive in an another.
     *
     */
    private void subscribe(){
        if(this.connection == CONNECTED){
            for(Subscription subscription : this.subscriptions.values()){
                Map<String, String> headers = new HashMap<String, String>();            
                headers.put(SUBSCRIPTION_ID, subscription.getId());
                headers.put(SUBSCRIPTION_DESTINATION, subscription.getDestination());

                subscribe(headers);
            }
        }
    }

    /**
     * Send the subscribe to the server with an header
     * @param headers
     *      header of a subscribe STOMP message
     */
    private void subscribe(Map<String, String> headers){
        transmit(COMMAND_SUBSCRIBE, headers, null);
    }

    /**
     * Destroy a subscription with its id
     * 
     * @param id
     *      the id of the subscription. This id is automatically setting up in the subscribe method
     */
    public void unsubscribe(String id){
        if(this.connection == CONNECTED){
            Map<String, String> headers = new HashMap<String, String>();
            headers.put(SUBSCRIPTION_ID, id);

            this.subscriptions.remove(id);
            this.transmit(COMMAND_UNSUBSCRIBE, headers, null);
        }
    }
}

This one is the Frame of a Stomp message :

这是 Stomp 消息的帧：

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class Frame {
//  private final static String CONTENT_LENGTH = "content-length";

    private String command;
    private Map<String, String> headers;
    private String body;

    /**
     * Constructor of a Frame object. All parameters of a frame can be instantiate
     * 
     * @param command
     * @param headers
     * @param body
     */
    public Frame(String command, Map<String, String> headers, String body){
        this.command = command;     
        this.headers = headers != null ? headers : new HashMap<String, String>();
        this.body = body != null ? body : "";
    }

    public String getCommand(){
        return command;
    }

    public Map<String, String> getHeaders(){
        return headers;
    }

    public String getBody(){
        return body;
    }

    /**
     * Transform a frame object into a String. This method is copied on the objective C one, in the MMPReactiveStompClient
     * library
     * @return a frame object convert in a String
     */
    private String toStringg(){
        String strLines = this.command;
        strLines += Byte.LF;
        for(String key : this.headers.keySet()){
            strLines += key + ":" + this.headers.get(key);
            strLines += Byte.LF;
        }
        strLines += Byte.LF;
        strLines += this.body;
        strLines += Byte.NULL;

        return strLines;
    }

    /**
     * Create a frame from a received message. This method is copied on the objective C one, in the MMPReactiveStompClient
     * library
     * 
     * @param data
     *  a part of the message received from network, which represented a frame
     * @return
     *  An object frame
     */
    public static Frame fromString(String data){
        List<String> contents = new ArrayList<String>(Arrays.asList(data.split(Byte.LF)));

        while(contents.size() > 0 && contents.get(0).equals("")){
            contents.remove(0);
        }

        String command = contents.get(0);
        Map<String, String> headers = new HashMap<String, String>();
        String body = "";

        contents.remove(0);
        boolean hasHeaders = false;
        for(String line : contents){
            if(hasHeaders){
                for(int i=0; i < line.length(); i++){
                    Character c = line.charAt(i);
                    if(!c.equals('
public class Subscription {

    private String id;

    private String destination;

    private ListenerSubscription callback;

    public Subscription(String destination, ListenerSubscription callback){
        this.destination = destination;
        this.callback = callback;
    }

    public String getId() {
        return id;
    }
    public void setId(String id) {
        this.id = id;
    }

    public String getDestination() {
        return destination;
    }

    public ListenerSubscription getCallback() {
        return callback;
    }
}
'))
                        body += c;
                }
            } else{
                if(line.equals("")){
                    hasHeaders = true;
                } else {
                    String[] header = line.split(":");
                    headers.put(header[0], header[1]);
                }
            }
        }
        return new Frame(command, headers, body);   
    }

//    No need this method, a single frame will be always be send because body of the message will never be excessive
//    /**
//     * Transform a message received from server in a Set of objects, named frame, manageable by java
//     * 
//     * @param datas
//     *        message received from network
//     * @return
//     *        a Set of Frame
//     */
//    public static Set<Frame> unmarshall(String datas){
//      String data;
//      String[] ref = datas.split(Byte.NULL + Byte.LF + "*");//NEED TO VERIFY THIS PARAMETER
//      Set<Frame> results = new HashSet<Frame>();
//      
//      for (int i = 0, len = ref.length; i < len; i++) {
//            data = ref[i];
//            
//            if ((data != null ? data.length() : 0) > 0){
//              results.add(unmarshallSingle(data));//"unmarshallSingle" is the old name method for "fromString"
//            }
//        }         
//      return results;
//    }

    /**
     * Create a frame with based fame component and convert them into a string
     * 
     * @param command
     * @param headers
     * @param body
     * @return  a frame object convert in a String, thanks to <code>toStringg()</code> method
     */
    public static String marshall(String command, Map<String, String> headers, String body){
        Frame frame = new Frame(command, headers, body);
        return frame.toStringg();
    }

    private class Byte {
        public static final String LF = "\n";
        public static final String NULL = "
public interface ListenerWSNetwork {
    public void onState(int state);
}

import java.util.Map;
public interface ListenerSubscription {
    public void onMessage(Map<String, String> headers, String body);
}
";
    }
}

This one is an object used to establish a subscription via the stomp protocol :

这是一个用于通过 stomp 协议建立订阅的对象：

private StompClient mStompClient;

 // ...

 mStompClient = Stomp.over(WebSocket.class, "ws://localhost:8080/app/hello/websocket");
 mStompClient.connect();

 mStompClient.topic("/topic/greetings").subscribe(topicMessage -> {
     Log.d(TAG, topicMessage.getPayload());
 });

 mStompClient.send("/app/hello", "My first STOMP message!");

 // ...

 mStompClient.disconnect();

At least, there are two interfaces used as the "Run" java class, to listen web socket network and a given subscription canal

至少，有两个接口用作“Run”java 类，用于侦听 Web 套接字网络和给定的订阅渠道

  classpath 'me.tatarka:gradle-retrolambda:3.2.0'

For more information, don't hesitate to ask me.

欲了解更多信息，请随时问我。

回答by Nickolay Savchenko

My implementation of STOMP protocol for android (or plain java) with RxJava https://github.com/NaikSoftware/StompProtocolAndroid. Tested on STOMP server with SpringBoot. Simple example (with retrolambda):

我使用 RxJava https://github.com/NaikSoftware/StompProtocolAndroid为 android（或纯 java）实现 STOMP 协议。使用 SpringBoot 在 STOMP 服务器上测试。简单示例（使用retrolambda）：

apply plugin: 'me.tatarka.retrolambda'

android {
       .............
    compileOptions {
        sourceCompatibility JavaVersion.VERSION_1_8
        targetCompatibility JavaVersion.VERSION_1_8
    }
}


  dependencies {
      ............................
    compile 'org.java-websocket:Java-WebSocket:1.3.0'
    compile 'com.github.NaikSoftware:StompProtocolAndroid:1.1.5'
}

Add the following classpath in project :

在项目中添加以下类路径：

public class MainActivity extends AppCompatActivity {
    private StompClient mStompClient;
    public static  final String TAG="StompClient";
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        Button view = (Button) findViewById(R.id.button);
        view.setOnClickListener(e->  new LongOperation().execute(""));


    }


        private class LongOperation extends AsyncTask<String, Void, String> {
        private StompClient mStompClient;
        String TAG="LongOperation";
        @Override
        protected String doInBackground(String... params) {

            mStompClient = Stomp.over(WebSocket.class, "ws://localhost:8080/app/hello/websocket");
            mStompClient.connect();

            mStompClient.topic("/topic/greetings").subscribe(topicMessage -> {
                Log.d(TAG, topicMessage.getPayload());
            });

            mStompClient.send("/app/hello", "My first STOMP message!").subscribe();


            mStompClient.lifecycle().subscribe(lifecycleEvent -> {
                switch (lifecycleEvent.getType()) {

                    case OPENED:
                        Log.d(TAG, "Stomp connection opened");
                        break;

                    case ERROR:
                        Log.e(TAG, "Error", lifecycleEvent.getException());
                        break;

                    case CLOSED:
                        Log.d(TAG, "Stomp connection closed");
                        break;
                }
            });
            return "Executed";
        }

        @Override
        protected void onPostExecute(String result) {

        }

    }
}

Add the following thing in your app build.gradle :

在您的应用 build.gradle 中添加以下内容：

<uses-permission android:name="android.permission.INTERNET" />

All working asynchronously! You can call connect()after subscribe()and send(), messages will be pushed to queue.

全部异步工作！您可以connect()在subscribe()和之后调用send()，消息将被推送到队列中。

Additional features:

附加的功能：

• extra HTTP headers for handshake query (for passing auth token or other)
• you can implement own transport for library, just implement interface ConnectionProvider
• subscribe connection lifecycle events (connected, closed, error)

• 用于握手查询的额外 HTTP 标头（用于传递身份验证令牌或其他）
• 您可以为库实现自己的传输，只需实现接口 ConnectionProvider
• 订阅连接生命周期事件（连接、关闭、错误）

For Example :

例如 ：

private void connection() {
        Map<String,String> headersSetup = new HashMap<String,String>();
        Stomp stomp = new Stomp(hostUrl, headersSetup, new ListenerWSNetwork() {
            @Override
            public void onState(int state) {
            }
        });
        stomp.connect();
        stomp.subscribe(new Subscription(testUrl, new ListenerSubscription() {
            @Override
            public void onMessage(Map<String, String> headers, String body) {
            }
        }));
    }

Add Internet permission in manifest.xml

在 manifest.xml 中添加 Internet 权限

handshake += "";

回答by horkavlna

Perfect solution Eperrin thank you. I would like fill the complete solution e.g. in your Activity/Service phase you call connectionmethod of course doesn't in MainThread.

完美的解决方案 Eperrin 谢谢。我想填写完整的解决方案，例如在您的活动/服务阶段，您调用的connection方法当然不在 MainThread 中。

handshake += "\r\n";

And be careful in websocket weberknecht libraryis bug in WebSocketHandshake classin method verifyServerHandshakeHeadersis on the line 124 is check only if(!headers.get("Connection").equals("Upgrade"))and when server send upgradeinstead of Upgradeyou get error connection failed: missing header field in server handshake: Connectionyou have to turn off ignore cases if(!headers.get("Connection").equalsIgnoreCase("Upgrade"))

并且在 websocket weberknecht 库中要小心是WebSocketHandshake 类中的错误verifyServerHandshakeHeaders在线 124 仅检查if(!headers.get("Connection").equals("Upgrade"))并且当服务器发送升级而不是升级你得到错误连接失败：服务器握手中缺少标头字段：连接你必须关闭忽略情况if(!headers.get("Connection").equalsIgnoreCase("Upgrade"))

回答by Evgenia

With weberknecht library I had an exception from Spring "A fragmented control frame was received but control frames may not be fragmented". Fixed it by clearing extra spaces: WebSocketHandshake.java

使用 weberknecht 库时，我遇到了 Spring 的异常“收到了碎片化的控制帧，但控制帧可能不会被碎片化”。通过清除多余的空格来修复它： WebSocketHandshake.java

##代码##

instead of:

代替：

##代码##