[Android] 실시간 데이터 처리를 위한 STOMP 사용하기 (Kotlin)

새로운 프로젝트를 하면서 실시간으로 갱신되는 데이터 처리가 필요해서 Web Socket 위에서 동작하는 문자 기반 메세징 프로토콜인 STOMP를 사용했습니다. 이번에는 이 STOMP를 프로젝트에 적용하면서 공부했던 내용들을 정리해보려고 합니다.

 

STOMP를 간단히 설명하면 pub/sub 구조를 이용하여 client와 서버가 통신하는 방법입니다. 기존의 Web Socket은 1:1 통신만이 가능했다면 STOMP는 메시지 브로커가 존재하여 1 : N의 관계를 가질 수 있습니다.

Message Broker(메시지 브로커)는 Publisher(송신자)로부터 전달받은 메시지를Subscriber(수신자)로 전달해주는 중간 역할이며 응용 소프트웨어 간에 메시지를 교환할 수 있게 합니다. 그리고 기존에는 WebSocketSession에 직접 메시지를 전달해야 했다면 STOMP를 사용하면 문자열을 구독함으로써 WebSocketSession을 직접 다루지 않고서도 고도화할 수 있습니다. 이는 개발자가 읽기 쉽고 사용하기 편리하다는 장점이 있습니다.

 

stomp over websocket message

 

구글링을 해보니 Android에서 STOMP를 쉽게 사용할 수 있게 오픈소스가 제공되고 있었습니다. (최선 버전 : 2020년 11월 4일)

 

GitHub - NaikSoftware/StompProtocolAndroid: STOMP protocol via WebSocket for Android

 

다만, 본 프로젝트는 Java 기반으로 작성이 되어있어서 Kotlin기반의 프로젝트에서는 일부 동작이 안되었습니다. 그래서 조금 더 찾아보니 Kotlin 기반으로 작성된 프로젝트가 다음과 같이 존재하였습니다. (최신 버전 : 2019년 1월 8일)

 

GitHub - bishoybasily/stomp: Stomp client for kotlin

 

저는 위 코드를 조금 가공하여 안드로이드 라이브러리로 만들어 사용했습니다. RxJava를 통해서 pub/sub 구조를 만들었고 OkHttp3을 통하여 WebSocket을 구현되었습니다.

 

실질적으로 사용하는 방법은 다음의 코드가 전부여서 이해하는 데는 어렵지 않았습니다.

lateinit var stompConnection: Disposable
lateinit var topic: Disposable

val url = "ws://example.com/endpoint"
val intervalMillis = 1000L
val client = OkHttpClient()

val stomp = StompClient(url, intervalMillis, client)

// connect
stompConnection = stomp.connect().subscribe {
    when (it.type) {
        Event.Type.OPENED -> {

        }
        Event.Type.CLOSED -> {

        }
        Event.Type.ERROR -> {

        }
    }
}

// subscribe
topic = stomp.subscribe("/destination").subscribe { Log.i(TAG, it) }

// unsubscribe
topic.dispose()

// send
stomp.send("/destination", "dummy message").subscribe {
    if (it) {
    }
}

// disconnect
stompConnection.dispose()

val url = "ws://example.com/endpoint"
val intervalMillis = 1000L
val client = OkHttpClient()

val stomp = StompClient(url, intervalMillis, client)

// connect
stompConnection = stomp.connect().subscribe {
    when (it.type) {
        Event.Type.OPENED -> {

        }
        Event.Type.CLOSED -> {

        }
        Event.Type.ERROR -> {

        }
    }
}

intervalMillis는 Stomp의 연결이 끊길 시 주기적으로 몇 초 간격으로 연결을 시도할 것인지를 지정해주는 변수입니다. stompConnection을 통해 구독을 하여 Event의 타입에 따라 Stomp의 생명 주기를 관리할 수 있습니다.

url 같은 경우 http는 ws, https는 wss로 대체합니다.

 

// subscribe
topic = stomp.subscribe("/destination").subscribe { Log.i(TAG, it) }

// unsubscribe
topic.dispose()

// send
stomp.send("/destination", "dummy message").subscribe {
    if (it) {
    }
}

// disconnect
stompConnection.dispose()

pub/sub 구조이기 때문에 기본적으로 RxJava에서 쓰는 것 같이 사용해주면 됩니다. 저는 여기서 서버에 Send 할 데이터는 따로 없었기 때문에 send 부분을 제외하고 사용하였습니다.

 

  topic = stomp.join("/destination").subscribe { it ->
            val responseData = JSONObject(it).getString("message")
                                                
            val modelList = Gson().fromJson<ArrayList<Model>>(
                    responseData, TypeToken.getParameterized(
                        MutableList::class.java,
                        Model::class.java
                    ).type)
	}

문자열로 오기 때문에 이를 object 형태로 가공해주기 위해 다음과 같이 Gson을 통해서 object로 변경해주었습니다.