@Test void IngestFromStream_IngestionPropertiesWithNullTable_IllegalArgumentException() { String data = "Name, Age, Weight, Height"; InputStream inputStream = new ByteArrayInputStream(Charset.forName("UTF-8").encode(data).array()); StreamSourceInfo streamSourceInfo = new StreamSourceInfo(inputStream); ingestionProperties = new IngestionProperties("database", null); assertThrows(IllegalArgumentException.class, () -> streamingIngestClient.ingestFromStream(streamSourceInfo, ingestionProperties), "Expected IllegalArgumentException to be thrown, but it didn't"); }
@Override public IngestionResult ingestFromStream(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(streamSourceInfo, "streamSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); streamSourceInfo.validate(); ingestionProperties.validate(); String format = getFormat(ingestionProperties); String mappingReference = getMappingReference(ingestionProperties, format); try { InputStream stream = (streamSourceInfo.getCompressionType() != null) ? streamSourceInfo.getStream() : compressStream(streamSourceInfo.getStream(), streamSourceInfo.isLeaveOpen()); log.debug("Executing streaming ingest."); this.streamingClient.executeStreamingIngest(ingestionProperties.getDatabaseName(), ingestionProperties.getTableName(), stream, null, format, mappingReference, !(streamSourceInfo.getCompressionType() == null || !streamSourceInfo.isLeaveOpen())); } catch (DataClientException | IOException e) { log.error(e.getMessage(), e); throw new IngestionClientException(e.getMessage(), e); } catch (DataServiceException e) { log.error(e.getMessage(), e); throw new IngestionServiceException(e.getMessage(), e); } log.debug("Stream was ingested successfully."); IngestionStatus ingestionStatus = new IngestionStatus(); ingestionStatus.status = OperationStatus.Succeeded; ingestionStatus.table = ingestionProperties.getTableName(); ingestionStatus.database = ingestionProperties.getDatabaseName(); return new IngestionStatusResult(ingestionStatus); }
StreamingIngestClient 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(); String format = getFormat(ingestionProperties); String mappingReference = getMappingReference(ingestionProperties, format); try { InputStream stream = (streamSourceInfo.getCompressionType() != null) ? streamSourceInfo.getStream() : compressStream(streamSourceInfo.getStream(), streamSourceInfo.isLeaveOpen()); log.debug("Executing streaming ingest."); this.streamingClient.executeStreamingIngest(ingestionProperties.getDatabaseName(), ingestionProperties.getTableName(), stream, null, format, mappingReference, !(streamSourceInfo.getCompressionType() == null || !streamSourceInfo.isLeaveOpen())); } catch (DataClientException | IOException e) { log.error(e.getMessage(), e); throw new IngestionClientException(e.getMessage(), e); } catch (DataServiceException e) { log.error(e.getMessage(), e); throw new IngestionServiceException(e.getMessage(), e); } log.debug("Stream was ingested successfully."); IngestionStatus ingestionStatus = new IngestionStatus(); ingestionStatus.status = OperationStatus.Succeeded; ingestionStatus.table = ingestionProperties.getTableName(); ingestionStatus.database = ingestionProperties.getDatabaseName(); return new IngestionStatusResult(ingestionStatus); } }
StreamingIngestClient 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(); String format = getFormat(ingestionProperties); String mappingReference = getMappingReference(ingestionProperties, format); try { InputStream stream = (streamSourceInfo.getCompressionType() != null) ? streamSourceInfo.getStream() : compressStream(streamSourceInfo.getStream(), streamSourceInfo.isLeaveOpen()); log.debug("Executing streaming ingest."); this.streamingClient.executeStreamingIngest(ingestionProperties.getDatabaseName(), ingestionProperties.getTableName(), stream, null, format, mappingReference, !(streamSourceInfo.getCompressionType() == null || !streamSourceInfo.isLeaveOpen())); } catch (DataClientException | IOException e) { log.error(e.getMessage(), e); throw new IngestionClientException(e.getMessage(), e); } catch (DataServiceException e) { log.error(e.getMessage(), e); throw new IngestionServiceException(e.getMessage(), e); } log.debug("Stream was ingested successfully."); IngestionStatus ingestionStatus = new IngestionStatus(); ingestionStatus.status = OperationStatus.Succeeded; ingestionStatus.table = ingestionProperties.getTableName(); ingestionStatus.database = ingestionProperties.getDatabaseName(); return new IngestionStatusResult(ingestionStatus); } StreamingIngestClient(ConnectionStringBuilder csb);   StreamingIngestClient(StreamingClient streamingClient); }
StreamingIngestClient 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(); String format = getFormat(ingestionProperties); String mappingReference = getMappingReference(ingestionProperties, format); try { InputStream stream = (streamSourceInfo.getCompressionType() != null) ? streamSourceInfo.getStream() : compressStream(streamSourceInfo.getStream(), streamSourceInfo.isLeaveOpen()); log.debug("Executing streaming ingest."); this.streamingClient.executeStreamingIngest(ingestionProperties.getDatabaseName(), ingestionProperties.getTableName(), stream, null, format, mappingReference, !(streamSourceInfo.getCompressionType() == null || !streamSourceInfo.isLeaveOpen())); } catch (DataClientException | IOException e) { log.error(e.getMessage(), e); throw new IngestionClientException(e.getMessage(), e); } catch (DataServiceException e) { log.error(e.getMessage(), e); throw new IngestionServiceException(e.getMessage(), e); } log.debug("Stream was ingested successfully."); IngestionStatus ingestionStatus = new IngestionStatus(); ingestionStatus.status = OperationStatus.Succeeded; ingestionStatus.table = ingestionProperties.getTableName(); ingestionStatus.database = ingestionProperties.getDatabaseName(); return new IngestionStatusResult(ingestionStatus); } 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 ingestFromStream(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(streamSourceInfo, "streamSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); streamSourceInfo.validate(); ingestionProperties.validate(); String format = getFormat(ingestionProperties); String mappingReference = getMappingReference(ingestionProperties, format); try { InputStream stream = (streamSourceInfo.getCompressionType() != null) ? streamSourceInfo.getStream() : compressStream(streamSourceInfo.getStream(), streamSourceInfo.isLeaveOpen()); log.debug("Executing streaming ingest."); this.streamingClient.executeStreamingIngest(ingestionProperties.getDatabaseName(), ingestionProperties.getTableName(), stream, null, format, mappingReference, !(streamSourceInfo.getCompressionType() == null || !streamSourceInfo.isLeaveOpen())); } catch (DataClientException | IOException e) { log.error(e.getMessage(), e); throw new IngestionClientException(e.getMessage(), e); } catch (DataServiceException e) { log.error(e.getMessage(), e); throw new IngestionServiceException(e.getMessage(), e); } log.debug("Stream was ingested successfully."); IngestionStatus ingestionStatus = new IngestionStatus(); ingestionStatus.status = OperationStatus.Succeeded; ingestionStatus.table = ingestionProperties.getTableName(); ingestionStatus.database = ingestionProperties.getDatabaseName(); return new IngestionStatusResult(ingestionStatus); } 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();  }