@Test public void testExtractObjectInfo() throws Exception { List<String> objectIds = Arrays.asList("ob1", "ob2", "ob3"); byte[][] entries = new byte[][] { "some data".getBytes(), "".getBytes(), "another data".getBytes() }; ByteArrayOutputStream headerOutputStream = new ByteArrayOutputStream(); JsonHandler.writeAsOutputStream(objectIds, headerOutputStream); byte[] header = headerOutputStream.toByteArray(); ByteArrayOutputStream multiplexedByteArrayOutputStream = new ByteArrayOutputStream(); MultiplexedStreamWriter multiplexedStreamWriter = new MultiplexedStreamWriter(multiplexedByteArrayOutputStream); multiplexedStreamWriter.appendEntry(header.length, new ByteArrayInputStream(header)); for (byte[] entry : entries) { multiplexedStreamWriter.appendEntry(entry.length, new ByteArrayInputStream(entry)); } multiplexedStreamWriter.appendEndOfFile(); InputStream multiplexedInputStream = spy(multiplexedByteArrayOutputStream.toInputStream()); MultiplexedStreamObjectInfoListenerThread multiplexedStreamObjectInfoListenerThread = new MultiplexedStreamObjectInfoListenerThread(multiplexedInputStream, DigestType.SHA512, objectIds); List<ObjectInfo> result = multiplexedStreamObjectInfoListenerThread.call(); assertThat(result).hasSize(objectIds.size()); for (int i = 0; i < objectIds.size(); i++) { String objectId = objectIds.get(i); ObjectInfo objectInfo = result.get(i); assertThat(objectInfo.getObjectId()).isEqualTo(objectId); assertThat(objectInfo.getSize()).isEqualTo(entries[i].length); Digest digest = new Digest(DigestType.SHA512); digest.update(entries[i]); assertThat(objectInfo.getDigest()).isEqualTo(digest.digestHex()); } verify(multiplexedInputStream, atLeastOnce()).close(); }
@Override public List<ObjectInfo> call() throws IOException, InvalidParseOperationException { try { return computeDigests(); } catch (Exception ex) { LOGGER.error("An error occurred during digestion computation of bulk transfer", ex); throw ex; } finally { IOUtils.closeQuietly(this.inputStream); } }
MultiplexedStreamObjectInfoListenerThread implements Callable<List<ObjectInfo>> { @Override public List<ObjectInfo> call() throws IOException, InvalidParseOperationException { try { return computeDigests(); } catch (Exception ex) { LOGGER.error("An error occurred during digestion computation of bulk transfer", ex); throw ex; } finally { IOUtils.closeQuietly(this.inputStream); } } }
MultiplexedStreamObjectInfoListenerThread implements Callable<List<ObjectInfo>> { @Override public List<ObjectInfo> call() throws IOException, InvalidParseOperationException { try { return computeDigests(); } catch (Exception ex) { LOGGER.error("An error occurred during digestion computation of bulk transfer", ex); throw ex; } finally { IOUtils.closeQuietly(this.inputStream); } } MultiplexedStreamObjectInfoListenerThread(InputStream inputStream, DigestType digestType,
        List<String> objectIds); }
MultiplexedStreamObjectInfoListenerThread implements Callable<List<ObjectInfo>> { @Override public List<ObjectInfo> call() throws IOException, InvalidParseOperationException { try { return computeDigests(); } catch (Exception ex) { LOGGER.error("An error occurred during digestion computation of bulk transfer", ex); throw ex; } finally { IOUtils.closeQuietly(this.inputStream); } } MultiplexedStreamObjectInfoListenerThread(InputStream inputStream, DigestType digestType,
        List<String> objectIds); @Override List<ObjectInfo> call(); }
MultiplexedStreamObjectInfoListenerThread implements Callable<List<ObjectInfo>> { @Override public List<ObjectInfo> call() throws IOException, InvalidParseOperationException { try { return computeDigests(); } catch (Exception ex) { LOGGER.error("An error occurred during digestion computation of bulk transfer", ex); throw ex; } finally { IOUtils.closeQuietly(this.inputStream); } } MultiplexedStreamObjectInfoListenerThread(InputStream inputStream, DigestType digestType,
        List<String> objectIds); @Override List<ObjectInfo> call();  }