@RunWithCustomExecutor @Test public void testTransferThread() throws Exception { InputStream is = mock(InputStream.class); Driver driver = mock(Driver.class); StorageOffer storageOffer = mock(StorageOffer.class); Connection connection = mock(Connection.class); doReturn("OfferId").when(storageOffer).getId(); doReturn(connection).when(driver).connect("OfferId"); StorageBulkPutResult result = mock(StorageBulkPutResult.class); doReturn(result).when(connection).bulkPutObjects(any(StorageBulkPutRequest.class)); MultiplexedStreamTransferThread multiplexedStreamTransferThread = new MultiplexedStreamTransferThread( 2, DataCategory.UNIT, Arrays.asList("ob1", "ob2"), is, 100L, driver, storageOffer, DigestType.SHA512); StorageBulkPutResult storageBulkPutResult = multiplexedStreamTransferThread.call(); ArgumentCaptor<StorageBulkPutRequest> storageBulkPutRequest = ArgumentCaptor.forClass(StorageBulkPutRequest.class); verify(connection).bulkPutObjects(storageBulkPutRequest.capture()); assertThat(storageBulkPutRequest.getValue().getObjectIds()).containsExactly("ob1", "ob2"); assertThat(storageBulkPutRequest.getValue().getDataStream()).isEqualTo(is); assertThat(storageBulkPutRequest.getValue().getSize()).isEqualTo(100L); assertThat(storageBulkPutRequest.getValue().getDigestType()).isEqualTo(DigestType.SHA512); assertThat(storageBulkPutRequest.getValue().getTenantId()).isEqualTo(2); assertThat(storageBulkPutResult).isSameAs(result); verify(connection).close(); verify(is).close(); }
@Override public StorageBulkPutResult call() throws Exception { try { return storeInOffer(); } catch (Exception ex) { LOGGER.error("An error occurred during bulk transfer to offer " + this.storageOffer.getId(), ex); throw ex; } finally { IOUtils.closeQuietly(this.inputStream); } }
MultiplexedStreamTransferThread implements Callable<StorageBulkPutResult> { @Override public StorageBulkPutResult call() throws Exception { try { return storeInOffer(); } catch (Exception ex) { LOGGER.error("An error occurred during bulk transfer to offer " + this.storageOffer.getId(), ex); throw ex; } finally { IOUtils.closeQuietly(this.inputStream); } } }
MultiplexedStreamTransferThread implements Callable<StorageBulkPutResult> { @Override public StorageBulkPutResult call() throws Exception { try { return storeInOffer(); } catch (Exception ex) { LOGGER.error("An error occurred during bulk transfer to offer " + this.storageOffer.getId(), ex); throw ex; } finally { IOUtils.closeQuietly(this.inputStream); } } MultiplexedStreamTransferThread(int tenantId,
        DataCategory dataCategory, List<String> objectIds, InputStream inputStream,
        long size, Driver driver, StorageOffer storageOffer, DigestType digestType); }
MultiplexedStreamTransferThread implements Callable<StorageBulkPutResult> { @Override public StorageBulkPutResult call() throws Exception { try { return storeInOffer(); } catch (Exception ex) { LOGGER.error("An error occurred during bulk transfer to offer " + this.storageOffer.getId(), ex); throw ex; } finally { IOUtils.closeQuietly(this.inputStream); } } MultiplexedStreamTransferThread(int tenantId,
        DataCategory dataCategory, List<String> objectIds, InputStream inputStream,
        long size, Driver driver, StorageOffer storageOffer, DigestType digestType); @Override StorageBulkPutResult call(); }
MultiplexedStreamTransferThread implements Callable<StorageBulkPutResult> { @Override public StorageBulkPutResult call() throws Exception { try { return storeInOffer(); } catch (Exception ex) { LOGGER.error("An error occurred during bulk transfer to offer " + this.storageOffer.getId(), ex); throw ex; } finally { IOUtils.closeQuietly(this.inputStream); } } MultiplexedStreamTransferThread(int tenantId,
        DataCategory dataCategory, List<String> objectIds, InputStream inputStream,
        long size, Driver driver, StorageOffer storageOffer, DigestType digestType); @Override StorageBulkPutResult call();  }