CbWebsocket.kt

@Component
class CoinbaseWebsocket @Autowired constructor(
        @Value("\${cbpro.websocket.baseurl}") val websocketUrl: String
) {

    private final val mapper = ObjectMapper()
    private final val logger = LoggerFactory.getLogger(javaClass)

    private final val client = ReactorNettyWebSocketClient()
    init { client.maxFramePayloadLength = Int.MAX_VALUE }

    fun connect(subscription: SubscriptionMessage): Flux<Message> {

        val emitter = EmitterProcessor.create<Message>()
        val subscriptionMessage = mapper.writeValueAsString(subscription)

        logger.info("Coinbase Endpoint will connect to $websocketUrl")

        val session = client.execute(URI(websocketUrl)) { session ->

            logger.info("Coinbase Session started ${session.handshakeInfo}")
            logger.info("Sending Subscription Message $subscriptionMessage")

            session.send(Mono.just(session.textMessage(subscriptionMessage)))
                    .thenMany(session.receive()
                            .map { it.payloadAsText }
                            .doOnNext { logger.trace(it) }
                            .map { JsonMessageReader.deserialize(it) }
                            .subscribeWith(emitter)).then()
        }

        return emitter.doOnSubscribe { session.subscribe() }
    }