-
Notifications
You must be signed in to change notification settings - Fork 81
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Cleanup to BarrageStreamGenerator and its related classes in antipati…
- Loading branch information
Showing
15 changed files
with
547 additions
and
613 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
760 changes: 272 additions & 488 deletions
760
...ons/barrage/src/main/java/io/deephaven/extensions/barrage/BarrageStreamGeneratorImpl.java
Large diffs are not rendered by default.
Oops, something went wrong.
55 changes: 55 additions & 0 deletions
55
.../barrage/src/main/java/io/deephaven/extensions/barrage/ChunkListInputStreamGenerator.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| // | ||
| // Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending | ||
| // | ||
| package io.deephaven.extensions.barrage; | ||
|
|
||
| import io.deephaven.chunk.Chunk; | ||
| import io.deephaven.chunk.ChunkType; | ||
| import io.deephaven.chunk.attributes.Values; | ||
| import io.deephaven.engine.rowset.RowSet; | ||
| import io.deephaven.extensions.barrage.chunk.ChunkInputStreamGenerator; | ||
| import io.deephaven.extensions.barrage.util.StreamReaderOptions; | ||
| import io.deephaven.util.SafeCloseable; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Arrays; | ||
| import java.util.List; | ||
|
|
||
| public class ChunkListInputStreamGenerator implements SafeCloseable { | ||
| private final List<ChunkInputStreamGenerator> generators; | ||
| private final ChunkInputStreamGenerator emptyGenerator; | ||
|
|
||
| public ChunkListInputStreamGenerator(Class<?> type, Class<?> componentType, List<Chunk<Values>> data, | ||
| ChunkType chunkType) { | ||
| // create an input stream generator for each chunk | ||
| ChunkInputStreamGenerator[] generators = new ChunkInputStreamGenerator[data.size()]; | ||
|
|
||
| long rowOffset = 0; | ||
| for (int i = 0; i < data.size(); ++i) { | ||
| final Chunk<Values> valuesChunk = data.get(i); | ||
| generators[i] = ChunkInputStreamGenerator.makeInputStreamGenerator(chunkType, type, componentType, | ||
| valuesChunk, rowOffset); | ||
| rowOffset += valuesChunk.size(); | ||
| } | ||
| this.generators = Arrays.asList(generators); | ||
| emptyGenerator = ChunkInputStreamGenerator.makeInputStreamGenerator( | ||
| chunkType, type, componentType, chunkType.getEmptyChunk(), 0); | ||
| } | ||
|
|
||
| public List<ChunkInputStreamGenerator> generators() { | ||
| return generators; | ||
| } | ||
|
|
||
| public ChunkInputStreamGenerator.DrainableColumn empty(StreamReaderOptions options, RowSet rowSet) | ||
| throws IOException { | ||
| return emptyGenerator.getInputStream(options, rowSet); | ||
| } | ||
|
|
||
| @Override | ||
| public void close() { | ||
| for (ChunkInputStreamGenerator generator : generators) { | ||
| generator.close(); | ||
| } | ||
| emptyGenerator.close(); | ||
| } | ||
| } |
54 changes: 54 additions & 0 deletions
54
...ns/barrage/src/main/java/io/deephaven/extensions/barrage/ConsecutiveDrainableStreams.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,54 @@ | ||
| // | ||
| // Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending | ||
| // | ||
| package io.deephaven.extensions.barrage; | ||
|
|
||
| import io.deephaven.extensions.barrage.util.DefensiveDrainable; | ||
| import io.deephaven.util.SafeCloseable; | ||
| import io.deephaven.util.datastructures.SizeException; | ||
| import org.jetbrains.annotations.NotNull; | ||
|
|
||
| import java.io.IOException; | ||
| import java.io.OutputStream; | ||
|
|
||
| public class ConsecutiveDrainableStreams extends DefensiveDrainable { | ||
| final DefensiveDrainable[] streams; | ||
|
|
||
| public ConsecutiveDrainableStreams(final @NotNull DefensiveDrainable... streams) { | ||
| this.streams = streams; | ||
| } | ||
|
|
||
| @Override | ||
| public int drainTo(final OutputStream outputStream) throws IOException { | ||
| int total = 0; | ||
| for (final DefensiveDrainable stream : streams) { | ||
| final int expected = total + stream.available(); | ||
| total += stream.drainTo(outputStream); | ||
| if (expected != total) { | ||
| throw new IllegalStateException("drained message drained wrong number of bytes"); | ||
| } | ||
| if (total < 0) { | ||
| throw new IllegalStateException("drained message is too large; exceeds Integer.MAX_VALUE"); | ||
| } | ||
| } | ||
| return total; | ||
| } | ||
|
|
||
| @Override | ||
| public int available() throws SizeException, IOException { | ||
| int total = 0; | ||
| for (final DefensiveDrainable stream : streams) { | ||
| total += stream.available(); | ||
| if (total < 0) { | ||
| throw new SizeException("drained message is too large; exceeds Integer.MAX_VALUE", total); | ||
| } | ||
| } | ||
| return total; | ||
| } | ||
|
|
||
| @Override | ||
| public void close() throws IOException { | ||
| SafeCloseable.closeAll(streams); | ||
| super.close(); | ||
| } | ||
| } |
44 changes: 44 additions & 0 deletions
44
.../barrage/src/main/java/io/deephaven/extensions/barrage/DrainableByteArrayInputStream.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| // | ||
| // Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending | ||
| // | ||
| package io.deephaven.extensions.barrage; | ||
|
|
||
| import io.deephaven.extensions.barrage.util.DefensiveDrainable; | ||
|
|
||
| import java.io.IOException; | ||
| import java.io.OutputStream; | ||
| import java.util.Objects; | ||
|
|
||
| public class DrainableByteArrayInputStream extends DefensiveDrainable { | ||
|
|
||
| private byte[] buf; | ||
| private final int offset; | ||
| private final int length; | ||
|
|
||
| public DrainableByteArrayInputStream(final byte[] buf, final int offset, final int length) { | ||
| this.buf = Objects.requireNonNull(buf); | ||
| this.offset = offset; | ||
| this.length = length; | ||
| } | ||
|
|
||
| @Override | ||
| public int available() { | ||
| if (buf == null) { | ||
| return 0; | ||
| } | ||
| return length; | ||
| } | ||
|
|
||
| @Override | ||
| public int drainTo(final OutputStream outputStream) throws IOException { | ||
| if (buf != null) { | ||
| try { | ||
| outputStream.write(buf, offset, length); | ||
| } finally { | ||
| buf = null; | ||
| } | ||
| return length; | ||
| } | ||
| return 0; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.