281 lines
12 KiB
Kotlin
281 lines
12 KiB
Kotlin
package chats.data.remote.signalr
|
|
|
|
import android.util.Log
|
|
import io.reactivex.rxjava3.core.Single
|
|
import com.microsoft.signalr.HubConnection
|
|
import com.microsoft.signalr.HubConnectionBuilder
|
|
import com.microsoft.signalr.HubConnectionState
|
|
import chats.data.remote.dto.ChatDto
|
|
import chats.data.remote.dto.MessageDto
|
|
import kotlinx.coroutines.flow.*
|
|
import kotlinx.coroutines.CoroutineScope
|
|
import kotlinx.coroutines.Dispatchers
|
|
import kotlinx.coroutines.launch
|
|
import javax.inject.Inject
|
|
import javax.inject.Singleton
|
|
|
|
import kotlinx.coroutines.delay
|
|
|
|
data class ReadMessagesRequest(
|
|
val chatId: String,
|
|
val lastReadMessageId: String,
|
|
val lastReadSequenceId: Int
|
|
)
|
|
|
|
data class MessagesReadEvent(
|
|
@com.google.gson.annotations.SerializedName("chatId", alternate = ["ChatId"]) val chatId: String? = null,
|
|
@com.google.gson.annotations.SerializedName("userId", alternate = ["UserId"]) val userId: String? = null,
|
|
@com.google.gson.annotations.SerializedName("lastReadSequenceId", alternate = ["LastReadSequenceId"]) val lastReadSequenceId: Int? = null
|
|
) {
|
|
val effectiveChatId: String get() = chatId ?: ""
|
|
val effectiveUserId: String get() = userId ?: ""
|
|
val effectiveLastReadSequenceId: Int get() = lastReadSequenceId ?: 0
|
|
}
|
|
|
|
data class ReactionEvent(
|
|
@com.google.gson.annotations.SerializedName("messageId", alternate = ["MessageId"]) val messageId: String? = null,
|
|
@com.google.gson.annotations.SerializedName("chatId", alternate = ["ChatId"]) val chatId: String? = null,
|
|
@com.google.gson.annotations.SerializedName("userId", alternate = ["UserId"]) val userId: String? = null,
|
|
@com.google.gson.annotations.SerializedName("username", alternate = ["Username", "UserName"]) val username: String? = null,
|
|
@com.google.gson.annotations.SerializedName("emoji", alternate = ["Emoji"]) val emoji: String? = null
|
|
)
|
|
|
|
enum class ConnectionStatus { CONNECTED, CONNECTING, DISCONNECTED }
|
|
|
|
@Singleton
|
|
class ChatHubClient @Inject constructor() {
|
|
private var hubConnection: HubConnection? = null
|
|
private val _events = MutableSharedFlow<ChatEvent>(extraBufferCapacity = 1024)
|
|
val events: SharedFlow<ChatEvent> = _events.asSharedFlow()
|
|
|
|
private val _status = MutableStateFlow(ConnectionStatus.DISCONNECTED)
|
|
val status: StateFlow<ConnectionStatus> = _status.asStateFlow()
|
|
|
|
private val scope = CoroutineScope(Dispatchers.IO)
|
|
private var lastBaseUrl: String? = null
|
|
private var lastToken: String? = null
|
|
|
|
fun connect(baseUrl: String, accessToken: String) {
|
|
if (hubConnection?.connectionState == HubConnectionState.CONNECTED) return
|
|
|
|
lastBaseUrl = baseUrl
|
|
lastToken = accessToken
|
|
_status.value = ConnectionStatus.CONNECTING
|
|
|
|
hubConnection = HubConnectionBuilder.create("${baseUrl}/hubs/chat")
|
|
.withAccessTokenProvider(Single.just(accessToken))
|
|
.build()
|
|
|
|
setupHandlers()
|
|
|
|
hubConnection?.onClosed { exception ->
|
|
Log.e("ChatHubClient", "Connection closed. Reconnecting...", exception)
|
|
_status.value = ConnectionStatus.DISCONNECTED
|
|
scope.launch {
|
|
delay(5000)
|
|
connect(baseUrl, accessToken)
|
|
}
|
|
}
|
|
|
|
scope.launch {
|
|
try {
|
|
hubConnection?.start()?.blockingAwait()
|
|
_status.value = ConnectionStatus.CONNECTED
|
|
Log.d("ChatHubClient", "SignalR Connected")
|
|
} catch (e: Exception) {
|
|
Log.e("ChatHubClient", "SignalR Connection Error", e)
|
|
_status.value = ConnectionStatus.DISCONNECTED
|
|
}
|
|
}
|
|
}
|
|
|
|
private fun setupHandlers() {
|
|
hubConnection?.let { conn ->
|
|
conn.on("new_message", { message: MessageDto ->
|
|
_events.tryEmit(ChatEvent.NewMessage(message))
|
|
}, MessageDto::class.java)
|
|
|
|
conn.on("message_edited", { messageId: String, chatId: String, content: String ->
|
|
_events.tryEmit(ChatEvent.MessageEdited(messageId, chatId, content))
|
|
}, String::class.java, String::class.java, String::class.java)
|
|
|
|
conn.on("message_deleted", { messageId: String, chatId: String ->
|
|
_events.tryEmit(ChatEvent.MessageDeleted(messageId, chatId))
|
|
}, String::class.java, String::class.java)
|
|
|
|
conn.on("messages_read", { data: MessagesReadEvent ->
|
|
_events.tryEmit(ChatEvent.MessagesRead(
|
|
data.effectiveChatId,
|
|
data.effectiveUserId,
|
|
data.effectiveLastReadSequenceId
|
|
))
|
|
}, MessagesReadEvent::class.java)
|
|
|
|
conn.on("user_typing", { data: ReactionEvent ->
|
|
_events.tryEmit(ChatEvent.UserTyping(data.chatId ?: "", data.userId ?: ""))
|
|
}, ReactionEvent::class.java)
|
|
|
|
conn.on("user_stopped_typing", { data: ReactionEvent ->
|
|
_events.tryEmit(ChatEvent.UserStoppedTyping(data.chatId ?: "", data.userId ?: ""))
|
|
}, ReactionEvent::class.java)
|
|
|
|
conn.on("user_online", { userId: String ->
|
|
_events.tryEmit(ChatEvent.UserOnline(userId))
|
|
}, String::class.java)
|
|
|
|
conn.on("new_chat", { chat: ChatDto ->
|
|
_events.tryEmit(ChatEvent.NewChat(chat))
|
|
}, ChatDto::class.java)
|
|
|
|
conn.on("reaction_added", { data: ReactionEvent ->
|
|
_events.tryEmit(ChatEvent.ReactionUpdated(
|
|
data.messageId ?: "",
|
|
data.chatId ?: "",
|
|
data.userId ?: "",
|
|
data.emoji ?: "",
|
|
isRemoved = false
|
|
))
|
|
}, ReactionEvent::class.java)
|
|
|
|
conn.on("reaction_removed", { data: ReactionEvent ->
|
|
_events.tryEmit(ChatEvent.ReactionUpdated(
|
|
data.messageId ?: "",
|
|
data.chatId ?: "",
|
|
data.userId ?: "",
|
|
data.emoji ?: "",
|
|
isRemoved = true
|
|
))
|
|
}, ReactionEvent::class.java)
|
|
|
|
// WebRTC Signaling Handlers
|
|
conn.on("call_incoming", { chatId: String, from: String, offer: String, callType: String ->
|
|
_events.tryEmit(ChatEvent.CallIncoming(chatId, from, offer, callType))
|
|
}, String::class.java, String::class.java, String::class.java, String::class.java)
|
|
|
|
conn.on("call_answered", { chatId: String, answer: String ->
|
|
_events.tryEmit(ChatEvent.CallAnswered(chatId, answer))
|
|
}, String::class.java, String::class.java)
|
|
|
|
conn.on("ice_candidate", { chatId: String, candidate: String ->
|
|
_events.tryEmit(ChatEvent.IceCandidateReceived(chatId, candidate))
|
|
}, String::class.java, String::class.java)
|
|
|
|
conn.on("call_ended", { chatId: String ->
|
|
_events.tryEmit(ChatEvent.CallEnded(chatId))
|
|
}, String::class.java)
|
|
conn.on("message_pinned", { data: Map<String, Any> ->
|
|
// The web version expects a message object, but here we might get a partial DTO or just IDs.
|
|
// Let's assume we get { chatId, message: MessageDto } based on web
|
|
// We'll trust the DTO mapping if possible, but SignalR java client is picky with nested objects in Maps.
|
|
// For simplicity, we might needs a dedicated DTO if it fails.
|
|
}, Map::class.java)
|
|
|
|
conn.on("message_pinned", { chatId: String, message: MessageDto ->
|
|
_events.tryEmit(ChatEvent.MessagePinned(chatId, message))
|
|
}, String::class.java, MessageDto::class.java)
|
|
|
|
conn.on("message_unpinned", { chatId: String, messageId: String ->
|
|
_events.tryEmit(ChatEvent.MessageUnpinned(chatId, messageId))
|
|
}, String::class.java, String::class.java)
|
|
}
|
|
}
|
|
|
|
fun disconnect() {
|
|
hubConnection?.stop()
|
|
_status.value = ConnectionStatus.DISCONNECTED
|
|
}
|
|
|
|
fun addReaction(messageId: String, chatId: String, emoji: String) {
|
|
if (hubConnection?.connectionState == HubConnectionState.CONNECTED) {
|
|
hubConnection?.invoke("add_reaction", mapOf(
|
|
"messageId" to messageId,
|
|
"chatId" to chatId,
|
|
"emoji" to emoji
|
|
))?.doOnError { Log.e("ChatHubClient", "add_reaction error", it) }
|
|
?.subscribe()
|
|
Log.d("ChatHubClient", "Invoked add_reaction: $emoji on $messageId")
|
|
} else {
|
|
Log.w("ChatHubClient", "Cannot add_reaction: Not connected")
|
|
}
|
|
}
|
|
|
|
fun removeReaction(messageId: String, chatId: String, emoji: String) {
|
|
if (hubConnection?.connectionState == HubConnectionState.CONNECTED) {
|
|
hubConnection?.invoke("remove_reaction", mapOf(
|
|
"messageId" to messageId,
|
|
"chatId" to chatId,
|
|
"emoji" to emoji
|
|
))?.doOnError { Log.e("ChatHubClient", "remove_reaction error", it) }
|
|
?.subscribe()
|
|
Log.d("ChatHubClient", "Invoked remove_reaction: $emoji on $messageId")
|
|
}
|
|
}
|
|
|
|
fun pinMessage(messageId: String, chatId: String) {
|
|
if (hubConnection?.connectionState == HubConnectionState.CONNECTED) {
|
|
hubConnection?.invoke("pin_message", mapOf(
|
|
"messageId" to messageId,
|
|
"chatId" to chatId
|
|
))?.doOnError { Log.e("ChatHubClient", "pin_message error", it) }
|
|
?.subscribe()
|
|
}
|
|
}
|
|
|
|
fun unpinMessage(messageId: String, chatId: String) {
|
|
if (hubConnection?.connectionState == HubConnectionState.CONNECTED) {
|
|
hubConnection?.invoke("unpin_message", mapOf(
|
|
"messageId" to messageId,
|
|
"chatId" to chatId
|
|
))?.doOnError { Log.e("ChatHubClient", "unpin_message error", it) }
|
|
?.subscribe()
|
|
}
|
|
}
|
|
|
|
fun readMessages(request: ReadMessagesRequest) {
|
|
if (hubConnection?.connectionState == HubConnectionState.CONNECTED) {
|
|
hubConnection?.invoke("read_messages", request)
|
|
?.doOnError { Log.e("ChatHubClient", "read_messages error", it) }
|
|
?.subscribe()
|
|
Log.d("ChatHubClient", "Sent read_messages for chat: ${request.chatId}")
|
|
}
|
|
}
|
|
|
|
fun joinChat(chatId: String) {
|
|
scope.launch {
|
|
// Wait for connection to be established if it's currently connecting
|
|
var attempts = 0
|
|
while (hubConnection?.connectionState != HubConnectionState.CONNECTED && attempts < 10) {
|
|
delay(500)
|
|
attempts++
|
|
}
|
|
|
|
if (hubConnection?.connectionState == HubConnectionState.CONNECTED) {
|
|
hubConnection?.invoke("join_chat", chatId)
|
|
?.doOnError { Log.e("ChatHubClient", "join_chat error", it) }
|
|
?.subscribe()
|
|
Log.d("ChatHubClient", "Joined chat room: $chatId")
|
|
} else {
|
|
Log.e("ChatHubClient", "Failed to join chat room $chatId: Not connected")
|
|
}
|
|
}
|
|
}
|
|
|
|
fun sendTypingIndicator(chatId: String) {
|
|
if (hubConnection?.connectionState == HubConnectionState.CONNECTED) {
|
|
hubConnection?.invoke("typing_start", chatId)
|
|
?.doOnError { Log.e("ChatHubClient", "typing_start error", it) }
|
|
?.subscribe()
|
|
Log.d("ChatHubClient", "Sent typing indicator for chat: $chatId")
|
|
}
|
|
}
|
|
|
|
fun sendUserStoppedTyping(chatId: String) {
|
|
if (hubConnection?.connectionState == HubConnectionState.CONNECTED) {
|
|
hubConnection?.invoke("typing_stop", chatId)
|
|
?.doOnError { Log.e("ChatHubClient", "typing_stop error", it) }
|
|
?.subscribe()
|
|
Log.d("ChatHubClient", "Sent user stopped typing for chat: $chatId")
|
|
}
|
|
}
|
|
}
|