안드로이드 채팅 앱 만들기 -2편 STOMP 소켓 연결로 실시간 채팅 구현

1편에서, RESTful API 를 이용한 방만들기, 방 조회 까지의 부분을 다뤘다.

이번 블로그에서는 STOMP를 이용하여 본격적으로 실시간 양방향 통신을 구현하는 방법에 대해 알아보려 한다.

build.gradle.kts(:app)

dependencies {
    ...
    // STOMP protocol client
    implementation("com.github.NaikSoftware:StompProtocolAndroid:1.6.6")

    // rx
    implementation("io.reactivex.rxjava2:rxjava:2.2.21")
    implementation("io.reactivex.rxjava2:rxandroid:2.1.1")
}

settings.gradle.kts

repositories {
    ...
    maven("https://jitpack.io")
}

ApiConstants

object ApiConstants {
    ...
    // WebSocket 엔드포인트
    const val WS_BASE_URL = "ws://192.168.0.34:9999/ws-chat-native"
    const val WS_BASE_URL_VD = "ws://10.0.2.2:9999/ws-chat-native" // 가상머신

    // STOMP 경로
    const val STOMP_PUBLISH_PREFIX = "/pub"
    const val STOMP_SUBSCRIBE_PREFIX = "/sub"
    ...
}

StompClientManager

/**
 *  STOMP 프로토콜을 사용한 웹소켓 연결 관리 클래스.
 */
@Singleton
class StompClientManager @Inject constructor() {
    private val TAG = "StompClientManager"

    private var stompClient: StompClient? = null
    // RxJava의 Disposable을 관리하기 위한 컴포지트
    private val compositeDisposable = CompositeDisposable()

    private val _connectionState = MutableStateFlow<ConnectionState>(ConnectionState.DISCONNECTED)
    val connectionState: StateFlow<ConnectionState> = _connectionState

    fun connect() {
        if (stompClient != null && stompClient?.isConnected == true) {
            return
        }

        val url = ApiConstants.getWebSocketUrl(ApiConstants.isEmulator)
        Log.d(TAG, "Connecting to WebSocket: $url")

        stompClient = Stomp.over(Stomp.ConnectionProvider.OKHTTP, url)

        val headers = listOf(
            StompHeader("Origin", "app//com.example.chatapp"),
            StompHeader("X-App-Package", "com.example.chatapp")
        )

        stompClient?.connect(headers)

        stompClient?.lifecycle()?.subscribe { lifecycleEvent ->
            when (lifecycleEvent.type) {
                LifecycleEvent.Type.OPENED -> {
                    _connectionState.value = ConnectionState.CONNECTED
                }
                LifecycleEvent.Type.CLOSED -> {
                    _connectionState.value = ConnectionState.DISCONNECTED
                }
                LifecycleEvent.Type.ERROR -> {
                    _connectionState.value = ConnectionState.ERROR
                }
                else -> {}
            }
        }?.let { compositeDisposable.add(it) } // Disposable 추가하여 메모리 누수 방지
    }

    fun disconnect() {
        stompClient?.disconnect()
        compositeDisposable.clear()
        _connectionState.value = ConnectionState.DISCONNECTED
    }

    fun subscribeToRoom(roomId: String, onMessageReceived: (ChatMessage) -> Unit): Disposable? {
        if (stompClient?.isConnected != true) {
            _connectionState.value = ConnectionState.ERROR
            return null
        }

        val topicDisposable = stompClient?.topic("$STOMP_SUBSCRIBE_PREFIX/$roomId")?.subscribe({ topicMessage ->
            val chatMessage = Gson().fromJson(topicMessage.payload, ChatMessage::class.java)
            onMessageReceived(chatMessage)
        }, { throwable ->
            Log.e(TAG, "Error subscribing to room: ${throwable.message}")
        })

        // Disposable 관리를 위해 CompositeDisposable에 추가
        topicDisposable?.let { compositeDisposable.add(it) }
        return topicDisposable
    }

    fun sendMessage(roomId: String, chatMessage: ChatMessage) {
        if (stompClient?.isConnected != true) {
            _connectionState.value = ConnectionState.ERROR
            return
        }

        stompClient?.send("${STOMP_PUBLISH_PREFIX}/$roomId", Gson().toJson(chatMessage))?.subscribe({
        }, { throwable ->
            Log.e("StompClientManager", "Error sending message: ${throwable.message}")
        })?.let { compositeDisposable.add(it) }
    }

    fun onDestroy() {
        disconnect()
    }
}

enum class ConnectionState {
    CONNECTED, DISCONNECTED, ERROR
}

ChatRepository

class ChatRepository @Inject constructor(
    private val stompClientManager: StompClientManager,
) {
    private val TAG = "ChatRepository"
    fun connectToSocket() {
        stompClientManager.connect()
    }

    fun disconnectFromSocket() {
        stompClientManager.disconnect()
    }

    val connectionState = stompClientManager.connectionState

    fun subscribeToRoom(roomId: String, onMessageReceived: (ChatMessage) -> Unit): Disposable? {
        Log.d(TAG, "Subscribing to room: $roomId")
        return stompClientManager.subscribeToRoom(roomId) { message ->
            Log.d(TAG, "Message received: $message")
            onMessageReceived(message)
        }
    }

    fun sendMessage(roomId: String, message: String, sender: String) {
        val chatMessage = ChatMessage(
            roomId = roomId,
            sender = sender,
            message = message
        )
        stompClientManager.sendMessage(roomId, chatMessage)
    }

    fun enterRoom(roomId: String, username: String) {
        val chatMessage = ChatMessage(
            roomId = roomId,
            sender = username,
            message = "$username has entered the room",
            type = MessageType.ENTER
        )
        stompClientManager.sendMessage(roomId, chatMessage)
    }

    fun leaveRoom(roomId: String, username: String) {
        val chatMessage = ChatMessage(
            roomId = roomId,
            sender = username,
            message = "$username has left the room",
            type = MessageType.LEAVE
        )
        stompClientManager.sendMessage(roomId, chatMessage)
    }
}

RoomViewModel

@HiltViewModel
class RoomViewModel @Inject constructor(
    private val chatRoomRepository: ChatRoomRepository,
    private val chatRepository: ChatRepository,
) : ViewModel() {

    private val _chatMessages = MutableStateFlow<List<ChatMessage>>(emptyList())
    val chatMessages: StateFlow<List<ChatMessage>> = _chatMessages

    private val _currentRoom = MutableStateFlow<ChatRoom?>(null)
    val currentRoom: StateFlow<ChatRoom?> = _currentRoom

    private val _messageText = mutableStateOf<String>("")
    val messageText: State<String> = _messageText

    private val _username = mutableStateOf<String>(UUID.randomUUID().toString())
    val username: State<String> = _username

    private val _isLoading = MutableStateFlow(false)
    val isLoading: StateFlow<Boolean> = _isLoading.asStateFlow()

    private val _error = MutableStateFlow<String?>(null)
    val error: StateFlow<String?> = _error.asStateFlow()

    val connectionState = chatRepository.connectionState

    private var messageSubscription: Disposable? = null

    init {
        connectToSocket()
    }

    fun connectToSocket() {
        chatRepository.connectToSocket()
    }

    fun disconnectFromSocket() {
        messageSubscription?.dispose()
        chatRepository.disconnectFromSocket()
    }

    fun updateMessageText(newText: String) {
        _messageText.value = newText
    }

    fun joinRoom(roomId: String, username: String) {
        viewModelScope.launch {
            try {
                val room = chatRoomRepository.getRoom(roomId)
                _currentRoom.value = room

                messageSubscription?.dispose()

                _chatMessages.value = emptyList()

                messageSubscription = chatRepository.subscribeToRoom(roomId) { message ->
                    viewModelScope.launch {
                        val currentMessages = _chatMessages.value
                        val newMessages = currentMessages + message
                        _chatMessages.value = newMessages

                        Log.d("RoomViewModel", "Updated messages list, size: ${newMessages.size}")
                    }
                }

                chatRepository.enterRoom(roomId, username)
            } catch (e: Exception) {
                Log.e("ChatRoomViewModel", "Error joining room: ${e.message}")
            }
        }
    }

    fun sendMessage(message: String, username: String) {
        _currentRoom.value?.let { room ->
            chatRepository.sendMessage(room.roomId, message, username)
        }
    }

    fun leaveRoom(username: String) {
        _currentRoom.value?.let { room ->
            chatRepository.leaveRoom(room.roomId, username)
            messageSubscription?.dispose()
            messageSubscription = null
            _currentRoom.value = null
            _chatMessages.value = emptyList()
        }
    }

    override fun onCleared() {
        super.onCleared()
        messageSubscription?.dispose()
        chatRepository.disconnectFromSocket()
    }
}