CbTickRecordingService.kt

@Component
class TickRecordingService
@Autowired constructor(val socket: CoinbaseWebsocket,
                       val tickRepository: TickRepository) {

    private val logger = LoggerFactory.getLogger(javaClass)

    fun recordProducts(products: Array<String>): Flux<List<Message>> {

        logger.info("Start Recording Tick Stream for Products: $products")
        
        val stream = socket.connect(SubscriptionMessage(products, arrayOf("ticker")))

        return stream.filter { it is Tick }
                .buffer(Duration.ofSeconds(3))
                .doOnNext { ticks -> tickRepository.store(ticks as List<Tick>) }
                .doOnError {
                    logger.error(it.message)
                    logger.error(it.stackTrace.toString())
                }

    }

}