lib/src/main/java/edu/utexas/tacc/tapis/files/lib/services/TransfersService.java
public Flux<TransferControlAction> streamControlMessages() {
        ConsumeOptions options = new ConsumeOptions();
        options.qos(1000);

        // Each new transfer needs its own queue. Autodelete them when
        // the connection closes.
        String queueName = "control." + UUID.randomUUID();
        QueueSpecification qspec = QueueSpecification.queue(queueName)
            .autoDelete(true);

        Flux<Delivery> controlMessageStream = receiver.consumeAutoAck(queueName, options);
        return controlMessageStream
            .delaySubscription(sender.declareQueue(qspec)
                .then(sender.bind(binding(CONTROL_EXCHANGE, "#", queueName)))
            ).flatMap(this::deserializeControlMessage);
    }
/**
     * Stream the messages coming off of the CONTROL_QUEUE
     *
     * @return A flux of ControlMessage
     */
Stream the messages coming off of the CONTROL_QUEUE
@return A flux of ControlMessage
Stream the messages coming off of the CONTROL_QUEUE @return A flux of ControlMessage
public Flux<TransferControlAction> streamControlMessages() {
        ConsumeOptions options = new ConsumeOptions();
        options.qos(1000);
        String queueName = "control." + UUID.randomUUID();
        QueueSpecification qspec = QueueSpecification.queue(queueName)
            .autoDelete(true);
        Flux<Delivery> controlMessageStream = receiver.consumeAutoAck(queueName, options);
        return controlMessageStream
            .delaySubscription(sender.declareQueue(qspec)
                .then(sender.bind(binding(CONTROL_EXCHANGE, "#", queueName)))
            ).flatMap(this::deserializeControlMessage);
    }
public Flux < TransferControlAction > streamControlMessages ( ) { ConsumeOptions options = new ConsumeOptions ( ) ; options . qos ( 1000 ) ; String queueName = "control." + UUID . randomUUID ( ) ; QueueSpecification qspec = QueueSpecification . queue ( queueName ) . autoDelete ( true ) ; Flux < Delivery > controlMessageStream = receiver . consumeAutoAck ( queueName , options ) ; return controlMessageStream . delaySubscription ( sender . declareQueue ( qspec ) . then ( sender . bind ( binding ( CONTROL_EXCHANGE , "#" , queueName ) ) ) ) . flatMap ( this :: deserializeControlMessage ) ; }
Stream the messages coming off of the CONTROL_QUEUE
 @return A flux of ControlMessage
Stream the messages coming off of the CONTROL_QUEUE @return A flux of ControlMessage
// Each new transfer needs its own queue. Autodelete them when // the connection closes.