class BulkPutTransferManager {

    BulkPutResult bulkSendDataToOffers(String workspaceContainerGUID, int tenantId,
        DataCategory dataCategory, List<String> offerIds, Map<String, Driver> storageDrivers,
        Map<String, StorageOffer> storageOffers, List<String> workspaceObjectURIs,
        List<String> objectIds) {

        ResultOrError<StreamAndInfo, BulkPutResult> streamWithInfoFromWorkspace = null;
        ResultOrError<StreamAndInfo, BulkPutResult> prependedStreamWithInfo = null;
        MultiplePipedInputStream streams = null;
        List<Future<StorageBulkPutResult>> transferThreadFutures = null;
        Future<List<ObjectInfo>> digestListenerFuture = null;

        try {

            // Get multiplexed stream from workspace
            streamWithInfoFromWorkspace = retrieveBulkDataFromWorkspace(workspaceContainerGUID, workspaceObjectURIs,
                offerIds);
            if (streamWithInfoFromWorkspace.hasError()) {
                return streamWithInfoFromWorkspace.getError();
            }

            // Prepend with header entry containing object Ids
            prependedStreamWithInfo =
                prependWithObjectIdsHeaderEntry(objectIds, streamWithInfoFromWorkspace.getResult(),
                    offerIds);
            if (prependedStreamWithInfo.hasError()) {
                return prependedStreamWithInfo.getError();
            }

            // We need 1 thread per offer + 1 thread for digest computing
            streams =
                new MultiplePipedInputStream(prependedStreamWithInfo.getResult().getStream(), offerIds.size() + 1);

            transferThreadFutures = startTransferThreads(tenantId, dataCategory, objectIds, offerIds, storageDrivers,
                storageOffers, streams, prependedStreamWithInfo.getResult().getSize());

            digestListenerFuture = startDigestComputeThread(offerIds, streams, objectIds);

            // Await termination with timeout
            long finalTimeout = transfertTimeoutHelper.getTransferTimeout(
                prependedStreamWithInfo.getResult().getSize());
            TimeoutStopwatch timeoutStopwatch = new TimeoutStopwatch(finalTimeout);

            ResultOrError<List<ObjectInfo>, BulkPutResult> objectInfos = awaitDigestListenerThread(digestListenerFuture,
                timeoutStopwatch, offerIds);
            if (objectInfos.hasError()) {
                return objectInfos.getError();
            }

            HashMap<String, OfferBulkPutStatus> statusByOfferIds = new HashMap<>();
            for (int rank = 0; rank < offerIds.size(); rank++) {

                String offerId = offerIds.get(rank);
                Future<StorageBulkPutResult> transferThreadFuture = transferThreadFutures.get(rank);

                OfferBulkPutStatus status =
                    awaitTransferThread(timeoutStopwatch, objectInfos.getResult(), offerId, transferThreadFuture);

                statusByOfferIds.put(offerId, status);
            }
            return new BulkPutResult(objectInfos.getResult(), statusByOfferIds);

        } finally {

            if (transferThreadFutures != null) {
                for (Future<StorageBulkPutResult> transferThreadFuture : transferThreadFutures) {
                    transferThreadFuture.cancel(true);
                }
            }

            if (digestListenerFuture != null) {
                digestListenerFuture.cancel(true);
            }

            if (streams != null) {
                streams.close();
            }
            if (prependedStreamWithInfo != null && prependedStreamWithInfo.hasResult()) {
                prependedStreamWithInfo.getResult().close();
            }
            if (streamWithInfoFromWorkspace != null && streamWithInfoFromWorkspace.hasResult()) {
                streamWithInfoFromWorkspace.getResult().close();
            }
        }
    }

      BulkPutTransferManager(WorkspaceClientFactory workspaceClientFactory, TransfertTimeoutHelper transfertTimeoutHelper);
    @VisibleForTesting  BulkPutTransferManager(WorkspaceClientFactory workspaceClientFactory, DigestType digestType,
        AlertService alertService, ExecutorService executor, TransfertTimeoutHelper transfertTimeoutHelper);

}