Re-use transform because it is no longer deprecated

Per discussion with @simonbasle

Author: acogoluegnes

src/main/java/reactor/rabbitmq/Receiver.java

@@ -80,7 +80,7 @@ public Receiver(ReceiverOptions options) {
             cm = options.getConnectionMonoConfigurator().apply(cm);
             cm = cm.doOnNext(conn -> connection.set(conn))
                     .subscribeOn(this.connectionSubscriptionScheduler)
-                    .composeNow(this::cache);
+                    .transform(this::cache);
         } else {
             cm = options.getConnectionMono();
         }

src/main/java/reactor/rabbitmq/Sender.java

@@ -110,7 +110,7 @@ public Sender(SenderOptions options) {
             cm = options.getConnectionMonoConfigurator().apply(cm);
             cm = cm.doOnNext(conn -> connection.set(conn))
                     .subscribeOn(this.connectionSubscriptionScheduler)
-                    .composeNow(this::cache);
+                    .transform(this::cache);
         } else {
             cm = options.getConnectionMono();
         }
@@ -124,7 +124,7 @@ public Sender(SenderOptions options) {
         this.resourceManagementScheduler = options.getResourceManagementScheduler() == null ?
             createScheduler("rabbitmq-sender-resource-creation") : options.getResourceManagementScheduler();
         this.resourceManagementChannelMono = options.getResourceManagementChannelMono() == null ?
-            connectionMono.map(CHANNEL_PROXY_CREATION_FUNCTION).composeNow(this::cache) : options.getResourceManagementChannelMono();
+            connectionMono.map(CHANNEL_PROXY_CREATION_FUNCTION).transform(this::cache) : options.getResourceManagementChannelMono();
         if (options.getConnectionClosingTimeout() != null && !Duration.ZERO.equals(options.getConnectionClosingTimeout())) {
             this.connectionClosingTimeout = (int) options.getConnectionClosingTimeout().toMillis();
         } else {
@@ -250,11 +250,11 @@ private BiConsumer<SignalType, Channel> getChannelCloseHandler(SendOptions optio
     }
 
     public RpcClient rpcClient(String exchange, String routingKey) {
-        return new RpcClient(connectionMono.map(CHANNEL_CREATION_FUNCTION).composeNow(this::cache), exchange, routingKey);
+        return new RpcClient(connectionMono.map(CHANNEL_CREATION_FUNCTION).transform(this::cache), exchange, routingKey);
     }
 
     public RpcClient rpcClient(String exchange, String routingKey, Supplier<String> correlationIdProvider) {
-        return new RpcClient(connectionMono.map(CHANNEL_CREATION_FUNCTION).composeNow(this::cache), exchange, routingKey, correlationIdProvider);
+        return new RpcClient(connectionMono.map(CHANNEL_CREATION_FUNCTION).transform(this::cache), exchange, routingKey, correlationIdProvider);
     }
 
     /**

src/test/java/reactor/rabbitmq/RabbitFluxTests.java

@@ -1107,7 +1107,7 @@ public void shovel() throws Exception {
                             sourceQueue,
                             new ConsumeOptions().stopConsumingBiFunction((emitter, msg) -> Integer.parseInt(new String(msg.getBody())) == nbMessages - 1)
                     ).map(delivery -> new OutboundMessage("", destinationQueue, delivery.getBody()))
-                            .composeNow(messages -> sender.send(messages)))
+                            .transform(messages -> sender.send(messages)))
                     .thenMany(receiver.consumeNoAck(destinationQueue)).subscribe(msg -> {
                         counter.incrementAndGet();
                         latch.countDown();