refactor(servicing): 重构推送消息处理机制

- 移除 PushMessageLiveData，改为使用 PushListener 接口
- 在 ServiceManager 中实现消息分发逻辑- 更新 PushMessageActivity 以接收广播消息
- 优化 JPushReceiver 和 MyMqttClient 的消息处理
- 调整 GlobalData 中的 isLoginRecognition 默认值
- 重构 SpeechManager 中的语音播放逻辑

servicing/build.gradle

@@ -73,7 +73,7 @@ publishing {
         release(MavenPublication) {
             groupId = 'io.github.szl9'
             artifactId = 'zd_servicing'
-            version = "1.0.1.9.9.12"
+            version = "1.0.1.9.9.25"
 
             pom {
                 packaging = "aar"

servicing/src/main/java/com/za/base/BaseActivityLifecycleCallbacks.kt

@@ -11,11 +11,13 @@ import java.util.concurrent.atomic.AtomicBoolean
 
 open class BaseActivityLifecycleCallbacks : ActivityLifecycleCallbacks {
 
+
 	override fun onActivityCreated(activity : Activity, savedInstanceState : Bundle?) {
 		LogUtil.print("onActivityCreated", activity.javaClass.simpleName)
 		keepScreenOn(activity)
 	}
 
+
 	override fun onActivityStarted(activity : Activity) {
 		LogUtil.print("onActivityStarted", activity.javaClass.simpleName)
 		synchronized(this) {
@@ -53,6 +55,7 @@ open class BaseActivityLifecycleCallbacks : ActivityLifecycleCallbacks {
 		}
 	}
 
+
 	override fun onActivitySaveInstanceState(activity : Activity, outState : Bundle) {
 		LogUtil.print("onActivitySaveInstanceState", activity.javaClass.simpleName)
 	}

servicing/src/main/java/com/za/base/PushMessageActivity.kt

@@ -4,15 +4,21 @@ import android.app.Activity
 import android.app.Notification
 import android.app.NotificationChannel
 import android.app.NotificationManager
+import android.content.BroadcastReceiver
 import android.content.Context
 import android.content.DialogInterface
+import android.content.Intent
+import android.content.IntentFilter
 import android.media.RingtoneManager
 import android.os.Build
 import android.os.Bundle
 import androidx.appcompat.app.AlertDialog
 import androidx.appcompat.app.AppCompatActivity
 import androidx.core.app.NotificationCompat
+import androidx.core.content.ContextCompat
+import com.blankj.utilcode.util.ActivityUtils
 import com.google.gson.Gson
+import com.za.base.BaseActivityLifecycleCallbacks.Companion.getCurrentActivity
 import com.za.bean.JpushBean
 import com.za.common.GlobalData
 import com.za.common.log.LogUtil
@@ -22,16 +28,103 @@ import com.za.ui.servicing.order_give_up.OrderGiveUpActivity
 import com.za.ui.view.CommonDialogFragment
 
 open class PushMessageActivity : AppCompatActivity() {
+	private var pushMessageReceiver : BroadcastReceiver? = null
 	private var currentDialog : AlertDialog? = null
 
 	override fun onCreate(savedInstanceState : Bundle?) {
 		super.onCreate(savedInstanceState)
-		if (! GlobalData.isMaster) {
+		if (! GlobalData.isMaster && pushMessageReceiver == null) {
-			PushMessageLiveData.pushMessage.observe(this) { message ->            // 处理推送消息
+			registerPushMessageReceiver(this)
-				LogUtil.print("PushMessageActivity", "Received push message: $message")
+		}
-				handlePushMessage(msg = message)
+	}
+
+	private fun registerPushMessageReceiver(context : Context) {
+		pushMessageReceiver = object : BroadcastReceiver() {
+			override fun onReceive(context : Context?, intent : Intent?) {
+				if (intent?.action == "com.za.rescue.dealer.PUSH_MESSAGE") {
+					val type = intent.getStringExtra("type") ?: return
+					val message = intent.getStringExtra("message") ?: return
+					LogUtil.print("PushActivityLifecycleCallbacks", "收到来自远程进程的消息: $type")
+					when (type) {
+						"broadcast" -> handleBroadcast("broadcast:$message")
+						"giveUp" -> {
+							try {
+								val jpushBean = Gson().fromJson(message, JpushBean::class.java)
+								val activity =
+									getCurrentActivity() ?: ActivityUtils.getTopActivity()
+								if (activity is AppCompatActivity) {
+									handleGiveUpOrder(activity, jpushBean)
+								}
+							} catch (e : Exception) {
+								LogUtil.print("PushActivityLifecycleCallbacks",
+									"处理订单放弃消息失败: ${e.message}")
+							}
+						}
+
+						"importantTip" -> {
+							try {
+								val jpushBean = Gson().fromJson(message, JpushBean::class.java)
+								val activity =
+									getCurrentActivity() ?: ActivityUtils.getTopActivity()
+								if (activity is AppCompatActivity) {
+									handleImportantTip(activity, jpushBean)
+								}
+							} catch (e : Exception) {
+								LogUtil.print("PushActivityLifecycleCallbacks",
+									"处理重要提示消息失败: ${e.message}")
+							}
+						}
+
+						"reDispatch" -> {
+							try {
+								val jpushBean = Gson().fromJson(message, JpushBean::class.java)
+								val activity =
+									getCurrentActivity() ?: ActivityUtils.getTopActivity()
+								if (activity is AppCompatActivity) {
+									handleReDispatchOrder(activity, jpushBean)
+								}
+							} catch (e : Exception) {
+								LogUtil.print("PushActivityLifecycleCallbacks",
+									"处理订单重新派发消息失败: ${e.message}")
+							}
+						}
+
+						"revoke" -> {
+							try {
+								handleRevokeOrder()
+							} catch (e : Exception) {
+								LogUtil.print("PushActivityLifecycleCallbacks",
+									"处理订单撤回消息失败: ${e.message}")
+							}
+						}
+					}
+				}
 			}
 		}
+
+		ContextCompat.registerReceiver(context,
+			pushMessageReceiver,
+			IntentFilter("com.za.rescue.dealer.PUSH_MESSAGE"),
+			ContextCompat.RECEIVER_NOT_EXPORTED)
+		LogUtil.print("PushActivityLifecycleCallbacks", "注册推送消息接收器")
+	}
+
+	override fun onDestroy() {
+		super.onDestroy()
+		if (isLastActivity()) {
+			try {
+				this.unregisterReceiver(pushMessageReceiver)
+				pushMessageReceiver = null
+				LogUtil.print("PushActivityLifecycleCallbacks", "注销推送消息接收器")
+			} catch (e : Exception) {
+				LogUtil.print("PushActivityLifecycleCallbacks",
+					"注销推送消息接收器失败: ${e.message}")
+			}
+		}
+	}
+
+	private fun isLastActivity() : Boolean {        // 检查是否是最后一个活动的Activity
+		return ActivityUtils.getActivityList().size <= 1
 	}
 
 	override fun onPause() {
@@ -39,36 +132,6 @@ open class PushMessageActivity : AppCompatActivity() {
 		dismissCurrentDialog()
 	}
 
-	private fun handlePushMessage(msg : String) {
-		if (msg.startsWith("broadcast:")) {
-			handleBroadcast(msg)
-			return
-		}
-		try {
-			val jpushOrderInfoBean = Gson().fromJson(msg, JpushBean::class.java)
-			when (jpushOrderInfoBean.pushType) {
-				1 -> handleTypeOneMessage(jpushOrderInfoBean)
-				3 -> handleImportantTip(jpushOrderInfoBean)
-				else -> LogUtil.print("JpushMessage",
-					"Unknown push type: ${jpushOrderInfoBean.pushType}")
-			}
-		} catch (e : Exception) {
-			if (msg.startsWith("broadcast:")) {
-				handleBroadcast(msg)
-			}
-			LogUtil.print("JpushMessage", "Error handling message: ${e.message}")
-		}
-	}
-
-	private fun handleTypeOneMessage(jpushOrderBean : JpushBean) {
-		when (jpushOrderBean.typeDesc) {
-			"giveUp" -> handleGiveUpOrder(jpushOrderBean)
-			"revoke" -> handleRevokeOrder()
-			"reDispatch" -> handleReDispatchOrder(jpushOrderBean)
-			else -> LogUtil.print("JpushMessage", "Unknown typeDesc: ${jpushOrderBean.typeDesc}")
-		}
-	}
-
 	// Handle broadcast messages
 	private fun handleBroadcast(msg : String) {
 		try {
@@ -90,7 +153,8 @@ open class PushMessageActivity : AppCompatActivity() {
 		}
 	}
 
-	private fun handleGiveUpOrder(jpushBean : JpushBean) {        // 播放提示音
+	private fun handleGiveUpOrder(activity : AppCompatActivity,
+		jpushBean : JpushBean) {        // 播放提示音
 		playNotificationSound(this)
 
 		if (GlobalData.currentOrder != null && GlobalData.currentOrder?.taskId == jpushBean.taskId) {
@@ -115,17 +179,16 @@ open class PushMessageActivity : AppCompatActivity() {
 		this.finish()
 	}
 
-	private fun handleReDispatchOrder(jpushBean : JpushBean) {
+	private fun handleReDispatchOrder(activity : AppCompatActivity, jpushBean : JpushBean) {
 		playNotificationSound(this)
 		currentDialog = AlertDialog.Builder(this).setTitle("订单重新派发")
 			.setMessage(buildReDispatchMessage(jpushBean)).setCancelable(false)
 			.setPositiveButton("确定") { dialog, _ ->
 				dialog.dismiss()
 			}.show()
-
 	}
 
-	private fun handleImportantTip(jpushBean : JpushBean) {
+	private fun handleImportantTip(activity : AppCompatActivity, jpushBean : JpushBean) {
 		playNotificationSound(this)
 		SpeechManager.speech("重要提醒：${jpushBean.tipContent ?: ""}")
 		currentDialog =

servicing/src/main/java/com/za/base/PushMessageLiveData.kt

@@ -1,59 +0,0 @@
-package com.za.base
-
-import android.util.Log
-import androidx.annotation.MainThread
-import androidx.lifecycle.LifecycleOwner
-import androidx.lifecycle.LiveData
-import androidx.lifecycle.MutableLiveData
-import androidx.lifecycle.Observer
-import cn.jpush.android.api.JPushMessage
-import java.util.concurrent.atomic.AtomicBoolean
-
-object PushMessageLiveData {
-	// 使用 SingleLiveEvent 或类似机制避免粘性事件问题
-	private val _pushMessage = SingleLiveEvent<String>()
-
-	// 对外暴露不可变的 LiveData
-	val pushMessage : LiveData<String> = _pushMessage
-
-	// 发送推送消息
-	fun postPushMessage(message : String) {
-		_pushMessage.postValue(message)
-	}
-}
-
-class SingleLiveEvent<T> : MutableLiveData<T>() {
-	private val pending = AtomicBoolean(false)
-
-	@MainThread
-	override fun observe(owner : LifecycleOwner, observer : Observer<in T>) {
-		if (hasActiveObservers()) {
-			Log.w(TAG, "多个观察者注册了 SingleLiveEvent，但只有一个会收到更新通知")
-		}
-
-		// 观察 LiveData 内部值
-		super.observe(owner) { t ->
-			if (pending.compareAndSet(true, false)) {
-				observer.onChanged(t)
-			}
-		}
-	}
-
-	@MainThread
-	override fun setValue(t : T?) {
-		pending.set(true)
-		super.setValue(t)
-	}
-
-	/**
-	 * 用于主线程外调用
-	 */
-	override fun postValue(value : T) {
-		pending.set(true)
-		super.postValue(value)
-	}
-
-	companion object {
-		private const val TAG = "SingleLiveEvent"
-	}
-}

servicing/src/main/java/com/za/common/GlobalData.kt

@@ -73,10 +73,10 @@ object GlobalData : GlobalLocalData() {
 	//是否已经完成登录后的人脸识别
 	var isLoginRecognition : Boolean? = null
 		get() {
-			return mmkv.decodeBool("isLoginRecognition", false)
+			return mmkv.decodeBool("isLoginRecognition", true)
 		}
 		set(value) {
-			mmkv.encode("isLoginRecognition", value ?: false)
+			mmkv.encode("isLoginRecognition", value ?: true)
 			field = value
 		}
 
@@ -117,7 +117,7 @@ object GlobalData : GlobalLocalData() {
 		currentLocation = null
 		driverInfoBean = null
 		loginTime = null
-		isLoginRecognition = null
+//		isLoginRecognition = null
 	}
 
 	fun clearAllOrderCache() {

servicing/src/main/java/com/za/common/speech/SpeechManager.kt

@@ -16,10 +16,7 @@ import com.za.room.db.user.LocalResourceBean
 import com.za.servicing.R
 import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
 import io.reactivex.rxjava3.schedulers.Schedulers
-import kotlinx.coroutines.CoroutineScope
+import kotlin.concurrent.thread
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.launch
 
 object SpeechManager {
 	private var mContext : Application? = null
@@ -182,7 +179,7 @@ object SpeechManager {
 		val localUrlResource = localResourceDao?.getLocalResourceByName(content)
 		if (localUrlResource != null && ! localUrlResource.resourceUrl.isNullOrBlank()) {
 			speechNewOrderLooper(content) {
-				playNewOrderFromNet(localUrlResource.resourceUrl ?: "")
+				playNewOrderFromNet(localUrlResource.resourceUrl)
 			}
 			LogUtil.print("handlerNewOrderVoice", "播放本地语音");
 			return
@@ -210,11 +207,11 @@ object SpeechManager {
 	}
 
 	private fun speechNewOrderLooper(content : String, play : () -> Unit) {
-		CoroutineScope(Dispatchers.IO).launch {
+		thread {
 			val startTime = System.currentTimeMillis()
 			while (System.currentTimeMillis() - startTime < 1000 * 60 * 3 && GlobalData.isHandlerNewOrder == false) {
 				play()
-				delay(250L * content.length)
+				Thread.sleep(250L * content.length)
 			}
 		}
 	}

servicing/src/main/java/com/za/service/ServiceManager.kt

@@ -4,13 +4,16 @@ import android.app.Notification
 import android.app.NotificationChannel
 import android.app.NotificationManager
 import android.content.Context
+import android.content.Intent
 import android.media.RingtoneManager
 import android.os.Build
+import android.os.Handler
+import android.os.Looper
 import androidx.core.app.NotificationCompat
 import cn.jiguang.api.utils.JCollectionAuth
 import cn.jpush.android.api.JPushInterface
+import com.blankj.utilcode.util.ProcessUtils
 import com.google.gson.Gson
-import com.za.base.PushMessageLiveData
 import com.za.bean.JpushBean
 import com.za.common.GlobalData
 import com.za.common.log.LogUtil
@@ -18,9 +21,20 @@ import com.za.common.util.DeviceUtil
 import com.za.service.mqtt.MyMqttClient
 import com.za.servicing.R
 
+interface PushListener {
+	fun newOrderMsg(jpushBean : JpushBean)
+	fun giveUpOrder(jpushBean : JpushBean)
+	fun revokeOrder(jpushBean : JpushBean)
+	fun reDispatchOrder(jpushBean : JpushBean)
+	fun broadcast(string : String)
+	fun importantTip(jpushBean : JpushBean)
+}
+
 data class LastJPushBean(val msg : Int, val time : Long = System.nanoTime())
 
 object ServiceManager {
+	@Volatile
+	private var pushListener : PushListener? = null
 	private var lastJPushBean : LastJPushBean? = null
 	private const val DUPLICATE_MSG_THRESHOLD = 3000L // 3秒
 
@@ -30,6 +44,49 @@ object ServiceManager {
 		LogUtil.print("ServiceManager", "Initializing ServiceManager")
 		jpushRegister(context)
 		MyMqttClient.initialize(deviceId = DeviceUtil.getAndroidId(context))
+		if (! GlobalData.isMaster) {
+			setupPushMessageReceiver(context)
+		}
+	}
+
+	private fun setupPushMessageReceiver(context : Context) {        // 注册推送消息接收器
+		registerPushListener(object : PushListener {
+			override fun broadcast(msg : String) {
+				sendMessageToMainProcess(context = context, "broadcast", msg)
+			}
+
+			override fun giveUpOrder(jpushBean : JpushBean) {
+				sendMessageToMainProcess(context = context, "giveUp", Gson().toJson(jpushBean))
+			}
+
+			override fun importantTip(jpushBean : JpushBean) {
+				sendMessageToMainProcess(context = context,
+					"importantTip",
+					Gson().toJson(jpushBean))
+			}
+
+			override fun newOrderMsg(jpushBean : JpushBean) {
+				sendMessageToMainProcess(context = context, "newOrder", Gson().toJson(jpushBean))
+			}
+
+			override fun reDispatchOrder(jpushBean : JpushBean) {
+				sendMessageToMainProcess(context = context, "reDispatch", Gson().toJson(jpushBean))
+			}
+
+			override fun revokeOrder(jpushBean : JpushBean) {
+				sendMessageToMainProcess(context = context, "revoke", Gson().toJson(jpushBean))
+			}
+		})
+	}
+
+	private fun sendMessageToMainProcess(context : Context,
+		type : String,
+		message : String) {        // 使用广播将消息发送到主进程
+		val intent = Intent("com.za.rescue.dealer.PUSH_MESSAGE")
+		intent.putExtra("type", type)
+		intent.putExtra("message", message)
+		context.sendBroadcast(intent)
+		LogUtil.print("KeepAliveService", "发送消息到主进程: $type")
 	}
 
 	// Register JPush
@@ -44,8 +101,15 @@ object ServiceManager {
 		}
 	}
 
+	// Register push listener
+	fun registerPushListener(listener : PushListener) {
+		Handler(Looper.getMainLooper()).post {
+			this.pushListener = listener
+			LogUtil.print("ServiceManager", "Registered push listener: ${this.pushListener}")
+		}
+	}
+
 	// Handle incoming push messages
-	@Synchronized
 	fun handlerPushMsg(msg : String) {
 		LogUtil.print("handlerPushMsg", "Received push message: $msg")
 
@@ -59,22 +123,96 @@ object ServiceManager {
 
 		if (msg.startsWith("broadcast:")) {
 			lastJPushBean = LastJPushBean(msg = msg.hashCode())
-			PushMessageLiveData.postPushMessage(msg)
+			handleBroadcast(msg) //			PushMessageLiveData.postPushMessage(msg)
 			return
 		}
 		try {
 			lastJPushBean = LastJPushBean(msg = msg.hashCode())
 			val jpushOrderInfoBean = Gson().fromJson(msg, JpushBean::class.java)
-			sendSystemNotificationFromMessage(jpushOrderInfoBean)
+			sendSystemNotificationFromMessage(jpushOrderInfoBean) //			PushMessageLiveData.postPushMessage(msg)
-			PushMessageLiveData.postPushMessage(msg)
+			when (jpushOrderInfoBean.pushType) {
+				0 -> newOrderMsg(jpushOrderInfoBean)
+				1 -> handleTypeOneMessage(jpushOrderInfoBean)
+				3 -> importantTip(jpushOrderInfoBean)
+				else -> LogUtil.print("JpushMessage",
+					"Unknown push type: ${jpushOrderInfoBean.pushType}")
+			}
 		} catch (e : Exception) {
-			if (msg.startsWith("broadcast:")) {
+			if (msg.startsWith("broadcast:")) { //				PushMessageLiveData.postPushMessage(msg)
-				PushMessageLiveData.postPushMessage(msg)
+				handleBroadcast(msg)
 			}
 			LogUtil.print("JpushMessage", "Error handling message: ${e.message}")
 		}
 	}
 
+	// Handle broadcast messages
+	private fun handleBroadcast(msg : String) {
+		try {
+			val content = msg.substring(10)
+			pushListener?.broadcast(content)
+			sendNotification(GlobalData.application, content)
+			LogUtil.print("JpushMessage", "Broadcast content: $content")
+		} catch (e : Exception) {
+			LogUtil.print("JpushMessage", "Broadcast failed: ${e.message}")
+		}
+	}
+
+	// Handle type one messages
+	private fun handleTypeOneMessage(jpushOrderBean : JpushBean) {
+		when (jpushOrderBean.typeDesc) {
+			"giveUp" -> giveUpOrder(jpushOrderBean)
+			"revoke" -> revokeOrder(jpushOrderBean)
+			"reDispatch" -> reDispatchOrder(jpushOrderBean)
+			else -> LogUtil.print("JpushMessage", "Unknown typeDesc: ${jpushOrderBean.typeDesc}")
+		}
+	}
+
+	// Handle new order messages
+	private fun newOrderMsg(jpushOrderBean : JpushBean) {
+		try {
+			LogUtil.print("JpushMessage",
+				"Handling new order message: ${this.pushListener} ${this.pushListener?.javaClass?.simpleName}")
+			LogUtil.print("isManThread", " ${ProcessUtils.getCurrentProcessName()}")
+			this.pushListener?.newOrderMsg(jpushOrderBean)
+		} catch (e : Exception) {
+			LogUtil.print("JpushMessage", "Failed to handle new order message: ${e.message}")
+		}
+	}
+
+	// Handle give up order messages
+	private fun giveUpOrder(jpushOrderBean : JpushBean) {
+		pushListener?.giveUpOrder(jpushOrderBean)
+	}
+
+	// Handle revoke order messages
+	private fun revokeOrder(jpushOrderBean : JpushBean) {
+		pushListener?.revokeOrder(jpushOrderBean)
+	}
+
+	// Handle re-dispatch order messages
+	private fun reDispatchOrder(jpushOrderBean : JpushBean) {
+		pushListener?.reDispatchOrder(jpushOrderBean)
+	}
+
+	// Handle important tip messages
+	private fun importantTip(jpushOrderBean : JpushBean) {
+		pushListener?.importantTip(jpushOrderBean)
+	}
+
+	// Disconnect from JPush and MQTT
+	fun disconnect(context : Context) {
+		Handler(Looper.getMainLooper()).post {
+			try {
+				JPushInterface.stopPush(context) // Stop JPush
+				MyMqttClient.disconnect() // Disconnect MQTT
+				LogUtil.print("ServiceManager", "Disconnected from JPush and MQTT successfully")
+			} catch (e : Exception) {
+				LogUtil.print("ServiceManager", "Error during disconnection: ${e.message}")
+			}
+		}
+	}
+
+
 	private const val CHANNEL_ID = "ImportantMessagesChannel"
 	private const val NOTIFICATION_ID = 1003
 
@@ -137,15 +275,4 @@ object ServiceManager {
 		notificationManager.notify(NOTIFICATION_ID, notification)
 	}
 
-	// Disconnect from JPush and MQTT
-	fun disconnect(context : Context) {
-		try {
-			JPushInterface.stopPush(context) // Stop JPush
-			MyMqttClient.disconnect() // Disconnect MQTT
-			LogUtil.print("ServiceManager", "Disconnected from JPush and MQTT successfully")
-		} catch (e : Exception) {
-			LogUtil.print("ServiceManager", "Error during disconnection: ${e.message}")
-		}
-	}
-
 }

servicing/src/main/java/com/za/service/jpush/JPushReceiver.kt

@@ -3,6 +3,8 @@ package com.za.service.jpush
 import android.content.BroadcastReceiver
 import android.content.Context
 import android.content.Intent
+import android.os.Handler
+import android.os.Looper
 import cn.jpush.android.api.JPushInterface
 import com.za.common.GlobalData
 import com.za.common.log.LogUtil
@@ -30,7 +32,11 @@ class JPushReceiver : BroadcastReceiver() {
 				if (msg.isNullOrBlank()) {
 					return
 				}
-				ServiceManager.handlerPushMsg(msg)
+
+				Handler(Looper.getMainLooper()).post {
+					LogUtil.print("JpushMessage ", "Received message: $msg")
+					ServiceManager.handlerPushMsg(msg)
+				}
 			}
 
 			JPushInterface.ACTION_NOTIFICATION_RECEIVED -> {

servicing/src/main/java/com/za/service/mqtt/MyMqttClient.kt

@@ -1,5 +1,7 @@
 package com.za.service.mqtt
 
+import android.os.Handler
+import android.os.Looper
 import android.util.Log
 import com.za.common.log.LogUtil
 import com.za.service.ServiceManager
@@ -54,9 +56,11 @@ object MyMqttClient {
 			}
 
 			override fun messageArrived(topic : String, mqttMessage : MqttMessage) {
-				val message = String(mqttMessage.payload)
+				Handler(Looper.getMainLooper()).post {
-				LogUtil.print("MyMqttClient ", "Message arrived: $message")
+					val message = String(mqttMessage.payload)
-				ServiceManager.handlerPushMsg(message) // Pass the message to ServiceManager for processing
+					LogUtil.print("MyMqttClient ", "Message arrived: $message")
+					ServiceManager.handlerPushMsg(message) // Pass the message to ServiceManager for processing
+				}
 			}
 
 			override fun deliveryComplete(token : IMqttDeliveryToken) {