@Test void IngestFromResultSet_IngestionPropertiesWithEmptyTable_IllegalArgumentException() { ResultSet resultSet = mock(ResultSet.class); ResultSetSourceInfo resultSetSourceInfo = new ResultSetSourceInfo(resultSet); ingestionProperties = new IngestionProperties("database", ""); assertThrows(IllegalArgumentException.class, () -> streamingIngestClient.ingestFromResultSet(resultSetSourceInfo, ingestionProperties), "Expected IllegalArgumentException to be thrown, but it didn't"); }
@Override public IngestionResult ingestFromResultSet(ResultSetSourceInfo resultSetSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(resultSetSourceInfo, "resultSetSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); resultSetSourceInfo.validate(); ingestionProperties.validate(); try { ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); new CsvRoutines().write(resultSetSourceInfo.getResultSet(), byteArrayOutputStream); byteArrayOutputStream.flush(); if (byteArrayOutputStream.size() <= 0) { String message = "Empty ResultSet."; log.error(message); throw new IngestionClientException(message); } ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(byteArrayOutputStream.toByteArray()); StreamSourceInfo streamSourceInfo = new StreamSourceInfo(byteArrayInputStream); return ingestFromStream(streamSourceInfo, ingestionProperties); } catch (IOException ex) { String msg = "Failed to read from ResultSet."; log.error(msg, ex); throw new IngestionClientException(msg, ex); } }
StreamingIngestClient implements IngestClient { @Override public IngestionResult ingestFromResultSet(ResultSetSourceInfo resultSetSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(resultSetSourceInfo, "resultSetSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); resultSetSourceInfo.validate(); ingestionProperties.validate(); try { ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); new CsvRoutines().write(resultSetSourceInfo.getResultSet(), byteArrayOutputStream); byteArrayOutputStream.flush(); if (byteArrayOutputStream.size() <= 0) { String message = "Empty ResultSet."; log.error(message); throw new IngestionClientException(message); } ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(byteArrayOutputStream.toByteArray()); StreamSourceInfo streamSourceInfo = new StreamSourceInfo(byteArrayInputStream); return ingestFromStream(streamSourceInfo, ingestionProperties); } catch (IOException ex) { String msg = "Failed to read from ResultSet."; log.error(msg, ex); throw new IngestionClientException(msg, ex); } } }
StreamingIngestClient implements IngestClient { @Override public IngestionResult ingestFromResultSet(ResultSetSourceInfo resultSetSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(resultSetSourceInfo, "resultSetSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); resultSetSourceInfo.validate(); ingestionProperties.validate(); try { ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); new CsvRoutines().write(resultSetSourceInfo.getResultSet(), byteArrayOutputStream); byteArrayOutputStream.flush(); if (byteArrayOutputStream.size() <= 0) { String message = "Empty ResultSet."; log.error(message); throw new IngestionClientException(message); } ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(byteArrayOutputStream.toByteArray()); StreamSourceInfo streamSourceInfo = new StreamSourceInfo(byteArrayInputStream); return ingestFromStream(streamSourceInfo, ingestionProperties); } catch (IOException ex) { String msg = "Failed to read from ResultSet."; log.error(msg, ex); throw new IngestionClientException(msg, ex); } } StreamingIngestClient(ConnectionStringBuilder csb);   StreamingIngestClient(StreamingClient streamingClient); }
StreamingIngestClient implements IngestClient { @Override public IngestionResult ingestFromResultSet(ResultSetSourceInfo resultSetSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(resultSetSourceInfo, "resultSetSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); resultSetSourceInfo.validate(); ingestionProperties.validate(); try { ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); new CsvRoutines().write(resultSetSourceInfo.getResultSet(), byteArrayOutputStream); byteArrayOutputStream.flush(); if (byteArrayOutputStream.size() <= 0) { String message = "Empty ResultSet."; log.error(message); throw new IngestionClientException(message); } ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(byteArrayOutputStream.toByteArray()); StreamSourceInfo streamSourceInfo = new StreamSourceInfo(byteArrayInputStream); return ingestFromStream(streamSourceInfo, ingestionProperties); } catch (IOException ex) { String msg = "Failed to read from ResultSet."; log.error(msg, ex); throw new IngestionClientException(msg, ex); } } 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 ingestFromResultSet(ResultSetSourceInfo resultSetSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(resultSetSourceInfo, "resultSetSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); resultSetSourceInfo.validate(); ingestionProperties.validate(); try { ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); new CsvRoutines().write(resultSetSourceInfo.getResultSet(), byteArrayOutputStream); byteArrayOutputStream.flush(); if (byteArrayOutputStream.size() <= 0) { String message = "Empty ResultSet."; log.error(message); throw new IngestionClientException(message); } ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(byteArrayOutputStream.toByteArray()); StreamSourceInfo streamSourceInfo = new StreamSourceInfo(byteArrayInputStream); return ingestFromStream(streamSourceInfo, ingestionProperties); } catch (IOException ex) { String msg = "Failed to read from ResultSet."; log.error(msg, ex); throw new IngestionClientException(msg, ex); } } 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();  }