Make fromFlux read streams eagerly (Azure#23907)

* Make fromFlux read streams eagerly

* Remove unused import

Author: srnagar

sdk/core/azure-core/src/main/java/com/azure/core/util/BinaryData.java

@@ -6,7 +6,6 @@
 import com.azure.core.util.implementation.BinaryDataContent;
 import com.azure.core.util.implementation.ByteArrayContent;
 import com.azure.core.util.implementation.FileContent;
-import com.azure.core.util.implementation.FluxByteBufferContent;
 import com.azure.core.util.implementation.InputStreamContent;
 import com.azure.core.util.implementation.SerializableContent;
 import com.azure.core.util.implementation.StringContent;
@@ -149,8 +148,7 @@ public static Mono<BinaryData> fromStreamAsync(InputStream inputStream) {
     }
 
     /**
-     * Creates an instance of {@link BinaryData} from the given {@link Flux} of {@link ByteBuffer}. The source flux
-     * is subscribed to as many times as the content is read. The flux, therefore, must be replayable.
+     * Creates an instance of {@link BinaryData} from the given {@link Flux} of {@link ByteBuffer}.
      *
      * <p><strong>Create an instance from a Flux of ByteBuffer</strong></p>
      *
@@ -164,12 +162,12 @@ public static Mono<BinaryData> fromFlux(Flux<ByteBuffer> data) {
         if (data == null) {
             return monoError(LOGGER, new NullPointerException("'content' cannot be null."));
         }
-        return Mono.just(new BinaryData(new FluxByteBufferContent(data)));
+        return FluxUtil.collectBytesInByteBufferStream(data)
+                .flatMap(bytes -> Mono.just(BinaryData.fromBytes(bytes)));
     }
 
     /**
-     * Creates an instance of {@link BinaryData} from the given {@link Flux} of {@link ByteBuffer}. The source flux
-     * is subscribed to as many times as the content is read. The flux, therefore, must be replayable.
+     * Creates an instance of {@link BinaryData} from the given {@link Flux} of {@link ByteBuffer}.
      *
      * <p><strong>Create an instance from a Flux of ByteBuffer</strong></p>
      *
@@ -185,10 +183,15 @@ public static Mono<BinaryData> fromFlux(Flux<ByteBuffer> data, Long length) {
         if (data == null) {
             return monoError(LOGGER, new NullPointerException("'content' cannot be null."));
         }
-        if (length < 0) {
+        if (length != null && length < 0) {
             return monoError(LOGGER, new IllegalArgumentException("'length' cannot be less than 0."));
         }
-        return Mono.just(new BinaryData(new FluxByteBufferContent(data, length)));
+        if (length != null) {
+            return FluxUtil.collectBytesInByteBufferStream(data, length.intValue())
+                    .flatMap(bytes -> Mono.just(BinaryData.fromBytes(bytes)));
+        }
+        return FluxUtil.collectBytesInByteBufferStream(data)
+                .flatMap(bytes -> Mono.just(BinaryData.fromBytes(bytes)));
     }
 
     /**