-
Notifications
You must be signed in to change notification settings - Fork 30
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #614 from gregschohn/NewAsyncBindings
- Loading branch information
Showing
42 changed files
with
1,367 additions
and
825 deletions.
There are no files selected for viewing
2 changes: 2 additions & 0 deletions
2
TrafficCapture/coreUtilities/src/main/java/org/opensearch/migrations/Utils.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
58 changes: 58 additions & 0 deletions
58
...ities/src/main/java/org/opensearch/migrations/utils/SequentialSpanCompressingReducer.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,58 @@ | ||
| package org.opensearch.migrations.utils; | ||
|
|
||
| import lombok.AllArgsConstructor; | ||
|
|
||
| /** | ||
| * This class can be used to reduce a stream of Integers into a string (calling getFinalAccumulation()) To use | ||
| * ``` | ||
| * Stream<Integer>...reduce(new SequentialSpanCompressingReducer(-1), SequentialSpanCompressingReducer::addNext, | ||
| * (c, d) -> { throw new IllegalStateException("parallel streams aren't allowed"); }) | ||
| * ``` | ||
| */ | ||
| @AllArgsConstructor | ||
| public class SequentialSpanCompressingReducer { | ||
| private static final int IGNORED_SENTINEL_VALUE = -1; | ||
| private static final char RUN_CHARACTER = '-'; | ||
|
|
||
| private final int shift; | ||
| private final int last; | ||
| private final StringBuilder accumulatedText; | ||
|
|
||
| public SequentialSpanCompressingReducer(int shift) { | ||
| this.shift = shift; | ||
| this.last = IGNORED_SENTINEL_VALUE; | ||
| this.accumulatedText = new StringBuilder(); | ||
| } | ||
|
|
||
| private boolean lastWasSpan() { | ||
| var len = accumulatedText.length(); | ||
| return len > 0 && accumulatedText.charAt(len-1) == RUN_CHARACTER; | ||
| } | ||
|
|
||
| public SequentialSpanCompressingReducer addNext(int b) { | ||
| if (last+shift == b) { | ||
| if (lastWasSpan()) { | ||
| return new SequentialSpanCompressingReducer(shift, b, | ||
| accumulatedText); | ||
| } else { | ||
| return new SequentialSpanCompressingReducer(shift, b, | ||
| accumulatedText.append(RUN_CHARACTER)); | ||
| } | ||
| } else { | ||
| if (lastWasSpan()) { | ||
| return new SequentialSpanCompressingReducer(shift, b, | ||
| accumulatedText.append(last).append(",").append(b)); | ||
| } else { | ||
| return new SequentialSpanCompressingReducer(shift, b, | ||
| accumulatedText.append(last == IGNORED_SENTINEL_VALUE ? "" : ",").append(b)); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public String getFinalAccumulation() { | ||
| if (lastWasSpan()) { | ||
| accumulatedText.append(last); | ||
| } | ||
| return accumulatedText.toString(); | ||
| } | ||
| } |
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
69 changes: 69 additions & 0 deletions
69
...icCapture/trafficReplayer/src/main/java/org/opensearch/migrations/NettyFutureBinders.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,69 @@ | ||
| package org.opensearch.migrations; | ||
|
|
||
| import io.netty.channel.EventLoop; | ||
| import io.netty.util.concurrent.Future; | ||
| import org.opensearch.migrations.replay.util.TrackedFuture; | ||
| import org.opensearch.migrations.replay.util.TextTrackedFuture; | ||
|
|
||
| import java.time.Duration; | ||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.function.Function; | ||
| import java.util.function.Supplier; | ||
|
|
||
| public class NettyFutureBinders { | ||
|
|
||
| private NettyFutureBinders() {} | ||
|
|
||
| public static CompletableFuture<Void> | ||
| bindNettyFutureToCompletableFuture(Future<?> nettyFuture, CompletableFuture<Void> cf) { | ||
| nettyFuture.addListener(f -> { | ||
| if (!f.isSuccess()) { | ||
| cf.completeExceptionally(f.cause()); | ||
| } else { | ||
| cf.complete(null); | ||
| } | ||
| }); | ||
| return cf; | ||
| } | ||
|
|
||
| public static CompletableFuture<Void> | ||
| bindNettyFutureToCompletableFuture(Future<?> nettyFuture) { | ||
| return bindNettyFutureToCompletableFuture(nettyFuture, new CompletableFuture<>()); | ||
| } | ||
|
|
||
| public static TrackedFuture<String, Void> | ||
| bindNettyFutureToTrackableFuture(Future<?> nettyFuture, String label) { | ||
| return new TextTrackedFuture<>(bindNettyFutureToCompletableFuture(nettyFuture), label); | ||
| } | ||
|
|
||
| public static TrackedFuture<String, Void> | ||
| bindNettyFutureToTrackableFuture(Future<?> nettyFuture, Supplier<String> labelProvider) { | ||
| return new TextTrackedFuture<>(bindNettyFutureToCompletableFuture(nettyFuture), labelProvider); | ||
| } | ||
|
|
||
| public static TrackedFuture<String, Void> | ||
| bindNettyFutureToTrackableFuture(Function<Runnable, Future<?>> nettyFutureGenerator, String label) { | ||
| return bindNettyFutureToTrackableFuture(nettyFutureGenerator.apply(() -> { | ||
| }), label); | ||
| } | ||
|
|
||
| public static TrackedFuture<String, Void> | ||
| bindNettySubmitToTrackableFuture(EventLoop eventLoop) { | ||
| return bindNettyFutureToTrackableFuture(eventLoop::submit, "waiting for event loop submission"); | ||
| } | ||
|
|
||
| public static TrackedFuture<String, Void> | ||
| bindNettyScheduleToCompletableFuture(EventLoop eventLoop, Duration delay) { | ||
| var delayMs = Math.max(0, delay.toMillis()); | ||
| return bindNettyFutureToTrackableFuture(eventLoop.schedule(() -> {}, delayMs, TimeUnit.MILLISECONDS), | ||
| "scheduling to run next send in " + delay + " (clipped: " + delayMs + "ms)"); | ||
| } | ||
|
|
||
| public static CompletableFuture<Void> | ||
| bindNettyScheduleToCompletableFuture(EventLoop eventLoop, Duration delay, CompletableFuture<Void> cf) { | ||
| var delayMs = Math.max(0, delay.toMillis()); | ||
| return bindNettyFutureToCompletableFuture(eventLoop.schedule(() -> { | ||
| }, delayMs, TimeUnit.MILLISECONDS), cf); | ||
| } | ||
| } |
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.