android 实现websocket 长链接功能

package demoapcx.baidumapsdk.myapplication; import android.os.Bundle; import android.support.v7.app.AppCompatActivity; import android.util.Log; import android.view.View; import android.widget.Toast; import com.google.gson.Gson; import com.google.gson.GsonBuilder; import io.reactivex.android.schedulers.AndroidSchedulers; import io.reactivex.disposables.CompositeDisposable; import io.reactivex.disposables.Disposable; import io.reactivex.schedulers.Schedulers; import ua.naiksoftware.stomp.Stomp; import ua.naiksoftware.stomp.StompClient; import ua.naiksoftware.stomp.dto.StompMessage; public class MainActivity extends AppCompatActivity { private static final String TAG = "MainActivity"; private StompClient mStompClient; private CompositeDisposable compositeDisposable; private Gson mGson = new GsonBuilder().create(); @Override protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.activity_main); mStompClient = Stomp.over(Stomp.ConnectionProvider.OKHTTP, "ws://192.168.103.60:8021"+"/websocket-address/websocket"); resetSubscriptions(); } /** * 关闭长链接 * @param view */ public void disconnectStomp(View view) { mStompClient.disconnect(); } /** * 开启长链接 * @param view */ public void connectStomp(View view) { mStompClient.withClientHeartbeat(1000).withServerHeartbeat(1000); resetSubscriptions(); Disposable dispLifecycle = mStompClient.lifecycle() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(lifecycleEvent -> { switch (lifecycleEvent.getType()) { case OPENED: toast("Stomp connection opened"); break; case ERROR: Log.e(TAG, "Stomp connection error", lifecycleEvent.getException()); toast("Stomp connection error"); break; case CLOSED: toast("Stomp connection closed"); resetSubscriptions(); break; case FAILED_SERVER_HEARTBEAT: toast("Stomp failed server heartbeat"); break; } }); compositeDisposable.add(dispLifecycle); // Receive greetings Disposable dispTopic = mStompClient.topic("/getLampPostAndLightCountByRoad") //Disposable dispTopic = mStompClient.topic("/topic") .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe((StompMessage topicMessage) -> { Log.d(TAG, "Received " + topicMessage.getPayload()); //DataModel response1 = mGson.fromJson(topicMessage.getPayload(), DataModel.class); //添加你的数据逻辑 }, throwable -> { Log.e(TAG, "连接错误", throwable); }); compositeDisposable.add(dispTopic); mStompClient.connect(); } /** * 长链接传递参数 * @param v */ public void sendEchoViaStomp(View v) { // if (!mStompClient.isConnected()) return; compositeDisposable.add(mStompClient.send("/topic/hello-msg-mapping", "Echo STOMP " ) .unsubscribeOn(Schedulers.newThread()) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(() -> { Log.d(TAG, "STOMP echo send successfully"); }, throwable -> { Log.e(TAG, "Error send STOMP echo", throwable); toast(throwable.getMessage()); })); } private void toast(String text) { Toast.makeText(this, text, Toast.LENGTH_SHORT).show(); } private void resetSubscriptions() { if (compositeDisposable != null) { compositeDisposable.dispose(); } compositeDisposable = new CompositeDisposable(); } @Override protected void onDestroy() { mStompClient.disconnect(); if (compositeDisposable != null) compositeDisposable.dispose(); super.onDestroy(); } }