@Test void IngestFromStream_NullStreamSourceInfo_IllegalArgumentException() { assertThrows( IllegalArgumentException.class, () -> queuedIngestClient.ingestFromStream(null, ingestionProperties)); }
@Override public IngestionResult ingestFromStream(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(streamSourceInfo, "streamSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); streamSourceInfo.validate(); ingestionProperties.validate(); try { IngestionResult ingestionResult; if (streamSourceInfo.getStream() == null) { throw new IngestionClientException("The provided stream is null."); } else if(streamSourceInfo.getStream().available() <= 0) { throw new IngestionClientException("The provided stream is empty."); } boolean shouldCompress = AzureStorageClient.shouldCompress(streamSourceInfo.getCompressionType(), ingestionProperties.getDataFormat()); String blobName = genBlobName( "StreamUpload", ingestionProperties.getDatabaseName(), ingestionProperties.getTableName(), ingestionProperties.getDataFormat(), shouldCompress ? CompressionType.gz : streamSourceInfo.getCompressionType()); CloudBlockBlob blob = azureStorageClient.uploadStreamToBlob( streamSourceInfo.getStream(), blobName, resourceManager.getIngestionResource(ResourceManager.ResourceType.TEMP_STORAGE), shouldCompress ); String blobPath = azureStorageClient.getBlobPathWithSas(blob); BlobSourceInfo blobSourceInfo = new BlobSourceInfo( blobPath, 0); ingestionResult = ingestFromBlob(blobSourceInfo, ingestionProperties); if (!streamSourceInfo.isLeaveOpen()) { streamSourceInfo.getStream().close(); } return ingestionResult; } catch (IOException | URISyntaxException e) { throw new IngestionClientException("Failed to ingest from stream", e); } catch (StorageException e) { throw new IngestionServiceException("Failed to ingest from stream", e); } catch (IngestionServiceException e) { validateEndpointServiceType(); throw e; } }
QueuedIngestClient implements IngestClient { @Override public IngestionResult ingestFromStream(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(streamSourceInfo, "streamSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); streamSourceInfo.validate(); ingestionProperties.validate(); try { IngestionResult ingestionResult; if (streamSourceInfo.getStream() == null) { throw new IngestionClientException("The provided stream is null."); } else if(streamSourceInfo.getStream().available() <= 0) { throw new IngestionClientException("The provided stream is empty."); } boolean shouldCompress = AzureStorageClient.shouldCompress(streamSourceInfo.getCompressionType(), ingestionProperties.getDataFormat()); String blobName = genBlobName( "StreamUpload", ingestionProperties.getDatabaseName(), ingestionProperties.getTableName(), ingestionProperties.getDataFormat(), shouldCompress ? CompressionType.gz : streamSourceInfo.getCompressionType()); CloudBlockBlob blob = azureStorageClient.uploadStreamToBlob( streamSourceInfo.getStream(), blobName, resourceManager.getIngestionResource(ResourceManager.ResourceType.TEMP_STORAGE), shouldCompress ); String blobPath = azureStorageClient.getBlobPathWithSas(blob); BlobSourceInfo blobSourceInfo = new BlobSourceInfo( blobPath, 0); ingestionResult = ingestFromBlob(blobSourceInfo, ingestionProperties); if (!streamSourceInfo.isLeaveOpen()) { streamSourceInfo.getStream().close(); } return ingestionResult; } catch (IOException | URISyntaxException e) { throw new IngestionClientException("Failed to ingest from stream", e); } catch (StorageException e) { throw new IngestionServiceException("Failed to ingest from stream", e); } catch (IngestionServiceException e) { validateEndpointServiceType(); throw e; } } }
QueuedIngestClient implements IngestClient { @Override public IngestionResult ingestFromStream(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(streamSourceInfo, "streamSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); streamSourceInfo.validate(); ingestionProperties.validate(); try { IngestionResult ingestionResult; if (streamSourceInfo.getStream() == null) { throw new IngestionClientException("The provided stream is null."); } else if(streamSourceInfo.getStream().available() <= 0) { throw new IngestionClientException("The provided stream is empty."); } boolean shouldCompress = AzureStorageClient.shouldCompress(streamSourceInfo.getCompressionType(), ingestionProperties.getDataFormat()); String blobName = genBlobName( "StreamUpload", ingestionProperties.getDatabaseName(), ingestionProperties.getTableName(), ingestionProperties.getDataFormat(), shouldCompress ? CompressionType.gz : streamSourceInfo.getCompressionType()); CloudBlockBlob blob = azureStorageClient.uploadStreamToBlob( streamSourceInfo.getStream(), blobName, resourceManager.getIngestionResource(ResourceManager.ResourceType.TEMP_STORAGE), shouldCompress ); String blobPath = azureStorageClient.getBlobPathWithSas(blob); BlobSourceInfo blobSourceInfo = new BlobSourceInfo( blobPath, 0); ingestionResult = ingestFromBlob(blobSourceInfo, ingestionProperties); if (!streamSourceInfo.isLeaveOpen()) { streamSourceInfo.getStream().close(); } return ingestionResult; } catch (IOException | URISyntaxException e) { throw new IngestionClientException("Failed to ingest from stream", e); } catch (StorageException e) { throw new IngestionServiceException("Failed to ingest from stream", e); } catch (IngestionServiceException e) { validateEndpointServiceType(); throw e; } } QueuedIngestClient(ConnectionStringBuilder csb);   QueuedIngestClient(ResourceManager resourceManager);   QueuedIngestClient(ResourceManager resourceManager, AzureStorageClient azureStorageClient); }
QueuedIngestClient implements IngestClient { @Override public IngestionResult ingestFromStream(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(streamSourceInfo, "streamSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); streamSourceInfo.validate(); ingestionProperties.validate(); try { IngestionResult ingestionResult; if (streamSourceInfo.getStream() == null) { throw new IngestionClientException("The provided stream is null."); } else if(streamSourceInfo.getStream().available() <= 0) { throw new IngestionClientException("The provided stream is empty."); } boolean shouldCompress = AzureStorageClient.shouldCompress(streamSourceInfo.getCompressionType(), ingestionProperties.getDataFormat()); String blobName = genBlobName( "StreamUpload", ingestionProperties.getDatabaseName(), ingestionProperties.getTableName(), ingestionProperties.getDataFormat(), shouldCompress ? CompressionType.gz : streamSourceInfo.getCompressionType()); CloudBlockBlob blob = azureStorageClient.uploadStreamToBlob( streamSourceInfo.getStream(), blobName, resourceManager.getIngestionResource(ResourceManager.ResourceType.TEMP_STORAGE), shouldCompress ); String blobPath = azureStorageClient.getBlobPathWithSas(blob); BlobSourceInfo blobSourceInfo = new BlobSourceInfo( blobPath, 0); ingestionResult = ingestFromBlob(blobSourceInfo, ingestionProperties); if (!streamSourceInfo.isLeaveOpen()) { streamSourceInfo.getStream().close(); } return ingestionResult; } catch (IOException | URISyntaxException e) { throw new IngestionClientException("Failed to ingest from stream", e); } catch (StorageException e) { throw new IngestionServiceException("Failed to ingest from stream", e); } catch (IngestionServiceException e) { validateEndpointServiceType(); throw e; } } QueuedIngestClient(ConnectionStringBuilder csb);   QueuedIngestClient(ResourceManager resourceManager);   QueuedIngestClient(ResourceManager resourceManager, AzureStorageClient azureStorageClient); @Override IngestionResult ingestFromBlob(BlobSourceInfo blobSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromFile(FileSourceInfo fileSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromStream(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromResultSet(ResultSetSourceInfo resultSetSourceInfo, IngestionProperties ingestionProperties); @Override void close(); }
QueuedIngestClient implements IngestClient { @Override public IngestionResult ingestFromStream(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(streamSourceInfo, "streamSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); streamSourceInfo.validate(); ingestionProperties.validate(); try { IngestionResult ingestionResult; if (streamSourceInfo.getStream() == null) { throw new IngestionClientException("The provided stream is null."); } else if(streamSourceInfo.getStream().available() <= 0) { throw new IngestionClientException("The provided stream is empty."); } boolean shouldCompress = AzureStorageClient.shouldCompress(streamSourceInfo.getCompressionType(), ingestionProperties.getDataFormat()); String blobName = genBlobName( "StreamUpload", ingestionProperties.getDatabaseName(), ingestionProperties.getTableName(), ingestionProperties.getDataFormat(), shouldCompress ? CompressionType.gz : streamSourceInfo.getCompressionType()); CloudBlockBlob blob = azureStorageClient.uploadStreamToBlob( streamSourceInfo.getStream(), blobName, resourceManager.getIngestionResource(ResourceManager.ResourceType.TEMP_STORAGE), shouldCompress ); String blobPath = azureStorageClient.getBlobPathWithSas(blob); BlobSourceInfo blobSourceInfo = new BlobSourceInfo( blobPath, 0); ingestionResult = ingestFromBlob(blobSourceInfo, ingestionProperties); if (!streamSourceInfo.isLeaveOpen()) { streamSourceInfo.getStream().close(); } return ingestionResult; } catch (IOException | URISyntaxException e) { throw new IngestionClientException("Failed to ingest from stream", e); } catch (StorageException e) { throw new IngestionServiceException("Failed to ingest from stream", e); } catch (IngestionServiceException e) { validateEndpointServiceType(); throw e; } } QueuedIngestClient(ConnectionStringBuilder csb);   QueuedIngestClient(ResourceManager resourceManager);   QueuedIngestClient(ResourceManager resourceManager, AzureStorageClient azureStorageClient); @Override IngestionResult ingestFromBlob(BlobSourceInfo blobSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromFile(FileSourceInfo fileSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromStream(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromResultSet(ResultSetSourceInfo resultSetSourceInfo, IngestionProperties ingestionProperties); @Override void close(); static final String INGEST_PREFIX; static final String EXPECTED_SERVICE_TYPE; }