WebSocket临时提交

app/build.gradle

@@ -52,6 +52,7 @@ dependencies {
 
 //    implementation libs.mvvmhabit
     implementation libs.okhttps
+    implementation libs.okhttps.stomp
     implementation libs.log.interceptor
     implementation libs.autosize
     implementation libs.fastble

app/src/main/AndroidManifest.xml

@@ -2,6 +2,7 @@
 <manifest xmlns:android="http://schemas.android.com/apk/res/android"
     xmlns:tools="http://schemas.android.com/tools" >
 
+    <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
     <uses-permission android:name="android.permission.INTERNET" />
     <uses-permission android:name="android.permission.BLUETOOTH" />
     <uses-permission android:name="android.permission.BLUETOOTH_ADMIN" />

app/src/main/java/com/grkj/iscs/extentions/Context.kt

@@ -0,0 +1,18 @@
+package com.grkj.iscs.extentions
+
+import android.content.Context
+import androidx.lifecycle.Observer
+import com.grkj.iscs.util.NetManager
+
+/**
+ * 网络管理器
+ */
+fun Context.netManager() = NetManager.getInstance(applicationContext)
+
+fun Context.addNetObserver(observer: Observer<Boolean>) {
+    netManager().liveData.observeForever(observer)
+}
+
+fun Context.removeNetObserver(observer: Observer<Boolean>) {
+    netManager().liveData.removeObserver(observer)
+}

app/src/main/java/com/grkj/iscs/model/Constants.kt

@@ -3,4 +3,7 @@ package com.grkj.iscs.model
 object Constants {
     const val PERMISSION_REQUEST_CODE = 1
     const val BLE_LOCAL_NAME = "keyLock"
+
+    const val BASE_URL = "http://192.168.1.3:48080"
+    const val WEB_SOCKET = ""
 }

app/src/main/java/com/grkj/iscs/util/NetHttpManager.kt

@@ -9,6 +9,7 @@ import cn.zhxu.okhttps.HttpTask
 import cn.zhxu.okhttps.OkHttps
 import cn.zhxu.okhttps.gson.GsonMsgConvertor
 import cn.zhxu.okhttps.okhttp.OkHttpClientWrapper
+import com.grkj.iscs.model.Constants
 import com.grkj.iscs.model.Token
 import okhttp3.logging.HttpLoggingInterceptor
 import java.io.IOException
@@ -34,7 +35,7 @@ class NetHttpManager {
             .addMsgConvertor(GsonMsgConvertor())
 //            .baseUrl("${UrlConsts.BASE_URL_PROD}$DEFAULT_DOMAIN")
 //            .baseUrl(Environment.baseUrl(ctx))
-            .baseUrl("http://192.168.1.3:48080")
+            .baseUrl(Constants.BASE_URL)
             .responseListener { task: HttpTask<*>?, result: HttpResult? ->
                 if (result?.status != 200) {
 //                    LogUtil.d(

app/src/main/java/com/grkj/iscs/util/NetManager.kt

@@ -0,0 +1,101 @@
+package com.grkj.iscs.util
+
+import android.content.Context
+import android.net.ConnectivityManager
+import android.net.Network
+import android.net.NetworkRequest
+import android.os.Build
+import androidx.annotation.WorkerThread
+import androidx.lifecycle.MutableLiveData
+import androidx.lifecycle.Observer
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+class NetManager private constructor(ctx: Context) {
+
+    @Volatile
+    private var available = false
+
+    val liveData = MutableLiveData<Boolean>()
+
+    private val latch = CountDownLatch(1)
+
+    private val callback = object : ConnectivityManager.NetworkCallback() {
+
+        override fun onAvailable(network: Network) {
+            liveData.postValue(true)
+            available = true
+            latch.countDown()
+        }
+
+        override fun onUnavailable() {
+            liveData.postValue(false)
+            available = false
+            latch.countDown()
+        }
+
+        override fun onLost(network: Network) {
+            liveData.postValue(false)
+            available = false
+            latch.countDown()
+        }
+
+    }
+
+    init {
+        (ctx.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager?)
+            ?.run {
+                if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.N) {
+                    registerDefaultNetworkCallback(callback)
+                } else {
+                    registerNetworkCallback(NetworkRequest.Builder().build(), callback)
+                }
+            }
+    }
+
+    /**
+     * 当前是否有网络，不能在主线程使用
+     */
+    @WorkerThread
+    fun isAvailable() : Boolean {
+        liveData.value?.let {
+            return it
+        }
+        latch.await(100, TimeUnit.MILLISECONDS)
+        return available
+    }
+
+    companion object {
+
+        @Volatile
+        private var instance: NetManager? = null
+
+        fun getInstance(ctx: Context): NetManager {
+            return instance ?: synchronized(this) {
+                instance ?: NetManager(ctx)
+                    .also {
+                        instance = it
+                    }
+            }
+        }
+
+    }
+
+}
+
+abstract class OnceNetObserver : Observer<Boolean> {
+
+    private var observed = false
+
+    final override fun onChanged(available: Boolean) {
+        synchronized(this) {
+            if (available && !observed) {
+                onNetAvailable()
+                observed = true
+            }
+        }
+    }
+
+    abstract fun onNetAvailable()
+
+}

app/src/main/java/com/grkj/iscs/util/StompApi.kt

@@ -0,0 +1,65 @@
+package com.grkj.iscs.util
+
+import android.content.Context
+import cn.zhxu.okhttps.OkHttps
+import cn.zhxu.stomp.Header
+import cn.zhxu.stomp.Stomp
+import com.grkj.iscs.extentions.addNetObserver
+import com.grkj.iscs.extentions.removeNetObserver
+import com.grkj.iscs.model.Constants
+import com.grkj.iscs.util.log.LogUtil
+
+object StompApi {
+
+    fun refuseStomp(ctx: Context, callback: (stomp: Stomp, baseDestination: String) -> Unit) {
+//        stomp(ctx, UrlConsts.MQ_VHOST_REFUSE, "machine.${ctx.serialNo()}", callback)
+    }
+
+
+    private fun stomp(ctx: Context, vhost: String, destinationInfix: String, callback: (stomp: Stomp, baseDestination: String) -> Unit) {
+        Executor.runOnMain {
+            ctx.addNetObserver(object : OnceNetObserver() {
+                override fun onNetAvailable() {
+                    onNetAvailable(ctx, vhost, destinationInfix, callback)
+                    ctx.removeNetObserver(this)
+                }
+            })
+        }
+    }
+
+    private fun onNetAvailable(ctx: Context, vhost: String, destinationInfix: String, callback: (stomp: Stomp, baseDestination: String) -> Unit) {
+//        PlatformRepository(ctx).getPlatformId().observeForever { platformId ->
+//            if (platformId != null) {
+//                Authorization.requestToken(ctx) { token ->
+//                    if (token != null) {
+//                        connectStompServer(ctx, vhost, destinationInfix, platformId, token, callback)
+//                    }
+//                }
+//            } else {
+//                LogUtil.w("platformId = null")
+//            }
+//        }
+    }
+
+    private fun connectStompServer(ctx: Context, vhost: String, destinationInfix: String, platformId: String, token: String, callback: (stomp: Stomp, baseDestination: String) -> Unit) {
+        LogUtil.i("开始连接 STOMP 服务")
+        Stomp.over(OkHttps.webSocket(Constants.WEB_SOCKET).heatbeat(20, 20))
+            .setOnConnected {
+                callback(it, "/p-${platformId}-a.${destinationInfix}")
+            }
+            .setOnDisconnected {
+                if (it.code != 1000) {
+                    LogUtil.i("Websockt 已断开：${it.code}: ${it.reason}，10 秒后重连...")
+                    Executor.delayOnIO({
+                        stomp(ctx, vhost, destinationInfix, callback)
+                    }, 10000)
+                }
+            }
+            .connect(listOf(
+                Header("login", token),
+                Header("host", vhost)
+            ))
+    }
+
+
+}

app/src/main/java/com/grkj/iscs/util/StompManager.kt

@@ -0,0 +1,115 @@
+package com.grkj.iscs.util
+
+import androidx.lifecycle.MutableLiveData
+import cn.zhxu.okhttps.OkHttps
+import cn.zhxu.stomp.Header
+import cn.zhxu.stomp.Stomp
+import com.grkj.iscs.model.Constants
+import com.grkj.iscs.util.log.LogUtil
+import java.util.concurrent.Executors
+import java.util.concurrent.ScheduledExecutorService
+import java.util.concurrent.TimeUnit
+
+class StompManager {
+    var myStomp: Stomp?=null
+    var scheduledExecutorService:ScheduledExecutorService?=null
+    var cacheTime=0L
+    var cacheDisconnectTime=0L
+    var topicRefuseStr:String?=null
+    val serverData = MutableLiveData<String>()
+
+    companion object{
+        fun getInstance() = InstanceHelper.sSingle
+    }
+
+    object InstanceHelper{
+        val sSingle = StompManager()
+    }
+
+    fun startStomp(){
+        if (myStomp?.isConnected==true){return}
+        LogUtil.d("stomp连接开始!")
+        NetHttpManager.getInstance().requestTokenAndRefreshIfExpired {
+            if (!it.isNullOrBlank()){
+                myStomp=Stomp.over(OkHttps.webSocket(Constants.WEB_SOCKET).heatbeat(10,10))
+                    .setOnConnected {
+                        scheduledExecutorService?.shutdownNow()
+                        scheduledExecutorService=null
+                        LogUtil.d("stomp连接完成!")
+                        cacheTime=System.currentTimeMillis()
+                        if (cacheDisconnectTime>20*1000){
+//                            NetStateUtil.instance.setNetState(0)
+                        }
+                        topicRefuse()
+                    }
+                    .setOnDisconnected {
+                        LogUtil.d("stomp连接断开：${it.reason}")
+                        cacheDisconnectTime=System.currentTimeMillis()-cacheTime
+//                        NetStateUtil.instance.setNetState(2)
+                        reConnectStomp()
+                    }
+                    .setOnException {
+                        LogUtil.d("stomp连接异常：${it.toString()}")
+                        cacheDisconnectTime=System.currentTimeMillis()-cacheTime
+//                        NetStateUtil.instance.setNetState(2)
+                        reConnectStomp()
+                    }
+                    .setOnError {
+                        LogUtil.d("stomp连接错误：${it.toString()}")
+                        cacheDisconnectTime=System.currentTimeMillis()-cacheTime
+//                        NetStateUtil.instance.setNetState(2)
+                        reConnectStomp()
+                    }
+                    .connect(listOf(Header("login",it),Header("host","/eiotyun-kitchen-refuse")))
+            }else{
+                reConnectStomp()
+            }
+        }
+    }
+
+    fun topicRefuse(){
+//        FenBiDeNetApi.getInstance().getPlatformInfo {
+//            it?.let {
+//                topicRefuseStr?.let {
+//                    myStomp?.untopic(it)
+//                }
+//                topicRefuseStr="/p-${it.platformId}-a.machine.${AppUtils.getSerialNo(FenBiDeApplication.instance)}"
+//                //创建设备
+//                myStomp?.topic(topicRefuseStr){
+//                    EventBus.getDefault().post(StompCreatEvent(it.payload=="1",it.payload))
+//                }
+//                //配置参数
+//                myStomp?.topic("${topicRefuseStr}.refresh"){
+//                    try {
+//                        EventBus.getDefault().post(Gson().fromJson(it.payload, StompRefreshEvent::class.java))
+//                    }catch (e:Exception){
+//                        ToastUtils.tip(FenBiDeApplication.instance.getString(R.string.str_stomp_exception,it.payload))
+//                        LogUtil.d("refresh：${FenBiDeApplication.instance.getString(R.string.str_stomp_exception,it.payload)}",e)
+//                    }
+//                }
+//            }?:let {
+//                try {
+//                    Thread.sleep(10000)
+//                }catch (e:Exception){}
+//                topicRefuse()
+//            }
+//        }
+    }
+
+    fun stopStomp(){
+        topicRefuseStr?.let {
+            myStomp?.untopic(it)
+        }
+        myStomp?.disconnect(true)
+    }
+
+    fun reConnectStomp(){
+        scheduledExecutorService?:let {
+            scheduledExecutorService= Executors.newScheduledThreadPool(1)
+            scheduledExecutorService?.scheduleWithFixedDelay(Runnable {
+                startStomp()
+            },0,30000, TimeUnit.MILLISECONDS)
+        }
+    }
+
+}

gradle/libs.versions.toml

@@ -26,6 +26,7 @@ androidx-activity = { group = "androidx.activity", name = "activity", version.re
 androidx-constraintlayout = { group = "androidx.constraintlayout", name = "constraintlayout", version.ref = "constraintlayout" }
 mvvmhabit = { group = "com.github.goldze", name = "MVVMHabit", version.ref = "mvvmhabit" }
 okhttps = { group = "cn.zhxu", name = "okhttps-gson", version.ref = "okhttps" }
+okhttps-stomp = { group = "cn.zhxu", name = "okhttps-stomp", version.ref = "okhttps" }
 log-interceptor = { group = "com.squareup.okhttp3", name = "logging-interceptor", version.ref = "log-interceptor" }
 autosize = { group = "com.github.JessYanCoding", name = "AndroidAutoSize", version.ref = "autosize" }
 fastble = { group = "com.github.Jasonchenlijian", name = "FastBle", version.ref = "fastble" }