How to auto reconnect after Internet off/on

When I off the mobile internet then websocket disconnect. but it not able auto reconnect after switch on internet.

   class WebSocketViewModel @Inject constructor(
   private var mStompClient: StompClient,
   private val loadActiveUserUseCase: LoadActiveUserUseCase
  ) : ViewModel() {
  private var compositeDisposable: CompositeDisposable = CompositeDisposable()
  private val mGson = GsonBuilder().create()
  private val _webSocketMessageAcknowledge = MutableLiveData<MessageAcknowledgeBean>()
  val webSocketMessageAcknowledge: LiveData<Event<MessageAcknowledgeBean>>
  private val _webSocketActiveUser = MutableLiveData<ActiveUserBean>()
  val webSocketActiveUser: LiveData<Event<ActiveUserBean>>
  private val _webSocketTotalViews = MutableLiveData<Event<ActiveUserBean>>()
  val webSocketTotalViews = _webSocketTotalViews
 private val _webSocketTotalLikes = MutableLiveData<Event<ActiveUserBean>>()
 val webSocketTotalLikes = _webSocketTotalLikes
 private val _webSocketTotalShare = MutableLiveData<Event<ActiveUserBean>>()
 val webSocketTotalShare = _webSocketTotalShare
 private val _loadActiveUserResult = MutableLiveData<Result<UserMaster>>()
 val activeUser: LiveData<Event<UserMaster>>
 lateinit var showId: String
 lateinit var userNameStr: String
 var isNeedToOpenUserNameDialog = false
 lateinit var mLoggedInUserId: String

  init {
    webSocketActiveUser = Transformations.map(_webSocketActiveUser)
    {
        Event(it)
    }
    webSocketMessageAcknowledge = Transformations.map(_webSocketMessageAcknowledge)
    {
        Event(it)
    }
    activeUser = Transformations.map(_loadActiveUserResult) {
        when (it) {
            is Result.Success -> Event(it.data)
            else -> Event(UserMaster())
        }
    }
    loadUser()
    mStompClient.withClientHeartbeat(1000).withServerHeartbeat(1000)

 }

    /**
     * Connect to webSocket
  */
    fun connectToWebSocket() {
    resetSubscriptions()
    val dispLifecycle = mStompClient.lifecycle()
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe { lifecycleEvent: LifecycleEvent ->

            when (lifecycleEvent.type) {

                LifecycleEvent.Type.OPENED -> {
                    // "Stomp connection opened"
                    Log.v("WebSocket", "Connected")
                }
                LifecycleEvent.Type.ERROR -> {
                    //"Stomp connection error"

                    Log.v("WebSocket", "Error"+lifecycleEvent.exception)
                }
                LifecycleEvent.Type.CLOSED -> {
                    //"Stomp connection closed"
                    reConnectToWebSocket()
                    Log.v("WebSocket", "Connection Closed")
                }
                LifecycleEvent.Type.FAILED_SERVER_HEARTBEAT -> {
                    //"Stomp failed server heartbeat"

                    Log.v("WebSocket", "Failed Heartbeat")
                }
                else -> {
                    //stomp event ype null

                    Log.v("WebSocket", "Retry Heartbeat")
                }
            }
          }

      compositeDisposable.add(dispLifecycle)
      acknowledgeMessage()
      activeUserCount()
      totalViewsCount()
     totalLikeCount()
     totalShareCount()
     mStompClient.connect()

  }

  private fun acknowledgeMessage() {
     val disposableTopic = mStompClient.topic(WebSocketConstant.MESSAGE_ACK)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(
            { topicMessage: StompMessage ->
                val message = mGson.fromJson(
                    topicMessage.payload,
                    MessageAcknowledgeBean::class.java
                )
                Log.v("WebSocket", topicMessage.payload)
                _webSocketMessageAcknowledge.value = (message)
            }
        ) { throwable: Throwable? ->
            //Exception
            throwable?.printStackTrace()
        }

    compositeDisposable.add(disposableTopic)
}

private fun activeUserCount() {
    val disposableTopic = mStompClient.topic(WebSocketConstant.ACTIVE_USER_EVENT)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(
            { topicMessage: StompMessage ->
                val message = mGson.fromJson(
                    topicMessage.payload,
                    ActiveUserBean::class.java
                )
                Log.v("WebSocket", topicMessage.payload)
                _webSocketActiveUser.value = (message)
            }
        ) { throwable: Throwable? ->
            //Exception
            throwable?.printStackTrace()
        }

    compositeDisposable.add(disposableTopic)
}

private fun totalViewsCount() {
    val disposableTopic = mStompClient.topic(WebSocketConstant.TOTAL_VIEWS_EVENT)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(
            { topicMessage: StompMessage ->
                val message = mGson.fromJson(
                    topicMessage.payload,
                    ActiveUserBean::class.java
                )
                Log.v("WebSocket", topicMessage.payload)
                _webSocketTotalViews.value = Event(message)
            }
        ) { throwable: Throwable? ->
            throwable?.printStackTrace()
        }

    compositeDisposable.add(disposableTopic)
}

private fun totalLikeCount() {
    val disposableTopic = mStompClient.topic(WebSocketConstant.TOTAL_LIKES_EVENT)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(
            { topicMessage: StompMessage ->
                val message = mGson.fromJson(
                    topicMessage.payload,
                    ActiveUserBean::class.java
                )
                Log.v("WebSocket", topicMessage.payload)
                _webSocketTotalLikes.value = Event(message)
            }
        ) { throwable: Throwable? ->
            throwable?.printStackTrace()
        }

    compositeDisposable.add(disposableTopic)
}

private fun totalShareCount() {
    val disposableTopic = mStompClient.topic(WebSocketConstant.TOTAL_SHARES_EVENT)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(
            { topicMessage: StompMessage ->
                val message = mGson.fromJson(
                    topicMessage.payload,
                    ActiveUserBean::class.java
                )
                Log.v("WebSocket", topicMessage.payload)
                _webSocketTotalShare.value = Event(message)
            }
        ) { throwable: Throwable? ->
            throwable?.printStackTrace()
        }

    compositeDisposable.add(disposableTopic)
}

fun sendEchoViaStomp() {
    if (!this::showId.isInitialized)
        return
    val userDetailDisposable = mStompClient.send(
        WebSocketConstant.USER_DETAIL_EVENT,
        sendValues()
    )
        .compose(applySchedulers())
        .subscribe(
            {
                //Subscribe
            }
        ) { throwable: Throwable ->
            throwable.printStackTrace()

        }
    compositeDisposable.add(
        userDetailDisposable
    )
}

private fun applySchedulers(): CompletableTransformer {
    return CompletableTransformer { upstream: Completable ->
        upstream
            .unsubscribeOn(Schedulers.newThread())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
    }

}

private fun sendValues(showId: String = this.showId): String {
    userNameStr = getUserName()
    val data = JSONObject()
    if (::mLoggedInUserId.isInitialized)
        data.put("userId", mLoggedInUserId)
    data.put("showId", showId)
    data.put("connectUserName", userNameStr)
    return data.toString()
}

private fun getUserName(): String {
    val userMaster = activeUser.value?.peekContent()
    return if (userMaster?.isLoggedIn == true) {
        if (TextUtils.isEmpty(userMaster.name))
            "Guest" + guestId()
        else
            userMaster.name
    } else {
        "Guest" + guestId()
    }
}

private fun guestId(): String {
    return EzMallApplication.activeUser.uuid.toString().substring(0, 6)
}

   /**
   *
  */
     private fun resetSubscriptions() {
    compositeDisposable.dispose()
    compositeDisposable = CompositeDisposable()
    compositeDisposable.clear()
  }

/**
 *  Reset Connection
 */
  private fun reConnectToWebSocket() {
    connectToWebSocket()

// if (mStompClient != null && !mStompClient.isConnected) // mStompClient.reconnect() }

 fun destroyWebSocket() {
     if (mStompClient != null) {
        mStompClient.disconnect()
     }
     compositeDisposable.dispose()
 }

 fun loadUser(isNeedToOpenUserNameDialog: Boolean = false) {
    this.isNeedToOpenUserNameDialog = isNeedToOpenUserNameDialog
    loadActiveUserUseCase.invoke(Unit, _loadActiveUserResult)
 }

}

NaikSoftware / StompProtocolAndroid

How to auto reconnect after Internet off/on #189