@Test void IngestFromFile_CompressedJson() throws Exception { String resourcesDirectory = System.getProperty("user.dir") + "/src/test/resources/"; String path = resourcesDirectory + "testdata.json"; FileSourceInfo fileSourceInfo = new FileSourceInfo(path, new File(path).length()); ingestionProperties.setDataFormat(IngestionProperties.DATA_FORMAT.json); ingestionProperties.setIngestionMapping("JsonMapping", IngestionMapping.IngestionMappingKind.Json); OperationStatus status = streamingIngestClient.ingestFromFile(fileSourceInfo, ingestionProperties).getIngestionStatusCollection().get(0).status; assertEquals(status, OperationStatus.Succeeded); verify(streamingClientMock, atLeastOnce()).executeStreamingIngest(any(String.class), any(String.class), any(InputStream.class), isNull(), any(String.class), any(String.class), any(boolean.class)); }
@Override public IngestionResult ingestFromFile(FileSourceInfo fileSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(fileSourceInfo, "fileSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); fileSourceInfo.validate(); ingestionProperties.validate(); try { String filePath = fileSourceInfo.getFilePath(); File file = new File(filePath); if (file.length() == 0) { String message = "Empty file."; log.error(message); throw new IngestionClientException(message); } InputStream stream = new FileInputStream(filePath); StreamSourceInfo streamSourceInfo = new StreamSourceInfo(stream, false, fileSourceInfo.getSourceId()); streamSourceInfo.setCompressionType(AzureStorageClient.getCompression(filePath)); return ingestFromStream(streamSourceInfo, ingestionProperties); } catch (FileNotFoundException e) { log.error("File not found when ingesting a file.", e); throw new IngestionClientException("IO exception - check file path.", e); } }
StreamingIngestClient implements IngestClient { @Override public IngestionResult ingestFromFile(FileSourceInfo fileSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(fileSourceInfo, "fileSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); fileSourceInfo.validate(); ingestionProperties.validate(); try { String filePath = fileSourceInfo.getFilePath(); File file = new File(filePath); if (file.length() == 0) { String message = "Empty file."; log.error(message); throw new IngestionClientException(message); } InputStream stream = new FileInputStream(filePath); StreamSourceInfo streamSourceInfo = new StreamSourceInfo(stream, false, fileSourceInfo.getSourceId()); streamSourceInfo.setCompressionType(AzureStorageClient.getCompression(filePath)); return ingestFromStream(streamSourceInfo, ingestionProperties); } catch (FileNotFoundException e) { log.error("File not found when ingesting a file.", e); throw new IngestionClientException("IO exception - check file path.", e); } } }
StreamingIngestClient implements IngestClient { @Override public IngestionResult ingestFromFile(FileSourceInfo fileSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(fileSourceInfo, "fileSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); fileSourceInfo.validate(); ingestionProperties.validate(); try { String filePath = fileSourceInfo.getFilePath(); File file = new File(filePath); if (file.length() == 0) { String message = "Empty file."; log.error(message); throw new IngestionClientException(message); } InputStream stream = new FileInputStream(filePath); StreamSourceInfo streamSourceInfo = new StreamSourceInfo(stream, false, fileSourceInfo.getSourceId()); streamSourceInfo.setCompressionType(AzureStorageClient.getCompression(filePath)); return ingestFromStream(streamSourceInfo, ingestionProperties); } catch (FileNotFoundException e) { log.error("File not found when ingesting a file.", e); throw new IngestionClientException("IO exception - check file path.", e); } } StreamingIngestClient(ConnectionStringBuilder csb);   StreamingIngestClient(StreamingClient streamingClient); }
StreamingIngestClient implements IngestClient { @Override public IngestionResult ingestFromFile(FileSourceInfo fileSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(fileSourceInfo, "fileSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); fileSourceInfo.validate(); ingestionProperties.validate(); try { String filePath = fileSourceInfo.getFilePath(); File file = new File(filePath); if (file.length() == 0) { String message = "Empty file."; log.error(message); throw new IngestionClientException(message); } InputStream stream = new FileInputStream(filePath); StreamSourceInfo streamSourceInfo = new StreamSourceInfo(stream, false, fileSourceInfo.getSourceId()); streamSourceInfo.setCompressionType(AzureStorageClient.getCompression(filePath)); return ingestFromStream(streamSourceInfo, ingestionProperties); } catch (FileNotFoundException e) { log.error("File not found when ingesting a file.", e); throw new IngestionClientException("IO exception - check file path.", e); } } StreamingIngestClient(ConnectionStringBuilder csb);   StreamingIngestClient(StreamingClient streamingClient); @Override IngestionResult ingestFromFile(FileSourceInfo fileSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromBlob(BlobSourceInfo blobSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromResultSet(ResultSetSourceInfo resultSetSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromStream(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties); @Override void close(); }
StreamingIngestClient implements IngestClient { @Override public IngestionResult ingestFromFile(FileSourceInfo fileSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(fileSourceInfo, "fileSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); fileSourceInfo.validate(); ingestionProperties.validate(); try { String filePath = fileSourceInfo.getFilePath(); File file = new File(filePath); if (file.length() == 0) { String message = "Empty file."; log.error(message); throw new IngestionClientException(message); } InputStream stream = new FileInputStream(filePath); StreamSourceInfo streamSourceInfo = new StreamSourceInfo(stream, false, fileSourceInfo.getSourceId()); streamSourceInfo.setCompressionType(AzureStorageClient.getCompression(filePath)); return ingestFromStream(streamSourceInfo, ingestionProperties); } catch (FileNotFoundException e) { log.error("File not found when ingesting a file.", e); throw new IngestionClientException("IO exception - check file path.", e); } } StreamingIngestClient(ConnectionStringBuilder csb);   StreamingIngestClient(StreamingClient streamingClient); @Override IngestionResult ingestFromFile(FileSourceInfo fileSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromBlob(BlobSourceInfo blobSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromResultSet(ResultSetSourceInfo resultSetSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromStream(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties); @Override void close();  }