[geode] 01/02: BufferedResultSender to make function result sending feel better.

[jb...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jbarrett pushed a commit to branch wip/BufferedResultSender
in repository https://gitbox.apache.org/repos/asf/geode.git

commit a8b5451d8133164669782fb9cb1349670d4f8ea8
Author: Jacob Barrett <jb...@pivotal.io>
AuthorDate: Fri Aug 23 00:50:29 2019 -0700

    BufferedResultSender to make function result sending feel better.
---
 .../geode/cache/execute/BufferedResultSender.java  | 156 +++++++++++++++++++
 .../cache/execute/BufferedResultSenderTest.java    | 168 +++++++++++++++++++++
 2 files changed, 324 insertions(+)

diff --git a/geode-core/src/main/java/org/apache/geode/cache/execute/BufferedResultSender.java b/geode-core/src/main/java/org/apache/geode/cache/execute/BufferedResultSender.java
new file mode 100644
index 0000000..73f8483
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/cache/execute/BufferedResultSender.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.geode.cache.execute;
+
+import java.util.Collection;
+
+/**
+ * Provides a more natural {@link ResultSender} behavior when whether a result is last or not is
+ * unknown. All {@link #sendResult(Object)} calls are buffered such that when {@link #close()} is
+ * called
+ * the last item in the buffer is sent via the underlying {@link ResultSender#lastResult(Object)}.
+ *
+ * <p>
+ * Example:
+ *
+ * <pre>
+ * {@code
+ *   try (BufferedResultSender<SomeType> resultSender = new BufferedResultSender<>(context.getResultSender()) {
+ *     for (SomeType result : results) {
+ *       resultSender.sendResult(result);
+ *     }
+ *   }
+ * }
+ * </pre>
+ *
+ * This snippet will send all the results in the results collection to the underlying
+ * {@link ResultSender}.
+ * </p>
+ *
+ * @param <T> type of result to send
+ */
+public class BufferedResultSender<T> implements ResultSender<T>, AutoCloseable {
+
+  private enum State {
+    EMPTY, BUFFERING, CLOSED
+  }
+
+  private final ResultSender<? super T> resultSender;
+  private T buffer;
+  private State state = State.EMPTY;
+
+  public static <T> BufferedResultSender<T> buffered(ResultSender<? super T> resultSender) {
+    return new BufferedResultSender<>(resultSender);
+  }
+
+  public BufferedResultSender(ResultSender<? super T> resultSender) {
+    this.resultSender = resultSender;
+  }
+
+  /**
+   * Sends buffered result by executing {@link ResultSender#lastResult(Object)} on underlying
+   * {@link ResultSender}. Since this method comes from {@link AutoCloseable} you may use this
+   * {@link ResultSender}
+   * in a try with resource block.
+   */
+  @Override
+  public void close() {
+    if (state == State.CLOSED) {
+      return;
+    }
+
+    sendLastAndClose();
+  }
+
+  /**
+   * Flush any buffered result by executing {@link ResultSender#sendResult(Object)} on underlying
+   * {@link ResultSender} and buffer new result.
+   *
+   * @param oneResult to buffer for sending.
+   */
+  @Override
+  public void sendResult(T oneResult) {
+    if (state == State.CLOSED) {
+      return;
+    }
+
+    bufferResult(oneResult);
+  }
+
+  /**
+   * Flushes any buffered result and executes {@link ResultSender#lastResult(Object)} on underlying
+   * {@link ResultSender}. Provided for backwards compatibility with {@link ResultSender} API but
+   * should not be used directly.
+   *
+   * @param lastResult to send
+   * @deprecated Use {@link #sendResult(Object)} and {@link #close()}
+   */
+  @Override
+  @Deprecated
+  public void lastResult(T lastResult) {
+    sendResult(lastResult);
+    sendLastAndClose();
+  }
+
+  /**
+   * Flushes any buffered result and executes {@link ResultSender#sendException(Throwable)} on
+   * underlying
+   * {@link ResultSender}.
+   *
+   * @param throwable to send
+   */
+  @Override
+  public void sendException(final Throwable throwable) {
+    if (state == State.CLOSED) {
+      return;
+    }
+
+    flushBuffer();
+    resultSender.sendException(throwable);
+    buffer = null;
+    state = State.CLOSED;
+  }
+
+  public void sendAllResults(final Collection<? extends T> results) {
+    if (state == State.CLOSED) {
+      return;
+    }
+
+    for (final T result : results) {
+      bufferResult(result);
+    }
+  }
+
+  private void bufferResult(final T oneResult) {
+    flushBuffer();
+    buffer = oneResult;
+  }
+
+  private void sendLastAndClose() {
+    resultSender.lastResult(buffer);
+    buffer = null;
+    state = State.CLOSED;
+  }
+
+  private void flushBuffer() {
+    if (state == State.EMPTY) {
+      state = State.BUFFERING;
+    } else {
+      resultSender.sendResult(buffer);
+    }
+  }
+
+}
diff --git a/geode-core/src/test/java/org/apache/geode/cache/execute/BufferedResultSenderTest.java b/geode-core/src/test/java/org/apache/geode/cache/execute/BufferedResultSenderTest.java
new file mode 100644
index 0000000..457787d
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/cache/execute/BufferedResultSenderTest.java
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.geode.cache.execute;
+
+import static java.util.Arrays.asList;
+import static org.apache.geode.cache.execute.BufferedResultSender.buffered;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.verifyZeroInteractions;
+
+import org.junit.Test;
+
+public class BufferedResultSenderTest {
+
+  @SuppressWarnings("unchecked")
+  private final ResultSender<Number> mockResultSender = mock(ResultSender.class);
+
+  @Test
+  public void closeInvokesLastResult() {
+    final BufferedResultSender<Integer> resultSender = new BufferedResultSender<>(mockResultSender);
+
+    resultSender.close();
+
+    verify(mockResultSender).lastResult(null);
+    verifyNoMoreInteractions(mockResultSender);
+  }
+
+  @Test
+  public void closeAfterCloseIsNoop() {
+    final BufferedResultSender<Integer> resultSender = new BufferedResultSender<>(mockResultSender);
+
+    resultSender.close();
+    resultSender.close();
+
+    verify(mockResultSender).lastResult(null);
+    verifyNoMoreInteractions(mockResultSender);
+  }
+
+  @Test
+  public void autoCloseInvokesLastResult() {
+    try (final BufferedResultSender<Integer> resultSender = buffered(mockResultSender)) {
+      // need to do something here to keep static analyzer happy
+      assertThat(resultSender).isNotNull();
+    }
+
+    verify(mockResultSender).lastResult(null);
+    verifyNoMoreInteractions(mockResultSender);
+  }
+
+  @Test
+  public void closeFlushesBuffer() {
+    try (final BufferedResultSender<Integer> resultSender = buffered(mockResultSender)) {
+      resultSender.sendResult(1);
+      resultSender.sendResult(2);
+    }
+
+    verify(mockResultSender).sendResult(1);
+    verify(mockResultSender).lastResult(2);
+    verifyNoMoreInteractions(mockResultSender);
+  }
+
+  @Test
+  public void sendResultBuffersFirstResult() {
+    final BufferedResultSender<Integer> resultSender = new BufferedResultSender<>(mockResultSender);
+
+    resultSender.sendResult(1);
+
+    verifyZeroInteractions(mockResultSender);
+  }
+
+  @Test
+  public void sendResultSendsFirstBuffersSecondResult() {
+    final BufferedResultSender<Integer> resultSender = new BufferedResultSender<>(mockResultSender);
+
+    resultSender.sendResult(1);
+    resultSender.sendResult(2);
+
+    verify(mockResultSender).sendResult(1);
+    verifyNoMoreInteractions(mockResultSender);
+  }
+
+  @Test
+  public void sendResultAfterCloseIsNoop() {
+    final BufferedResultSender<Integer> resultSender = new BufferedResultSender<>(mockResultSender);
+
+    resultSender.close();
+    resultSender.sendResult(1);
+
+    verify(mockResultSender).lastResult(null);
+    verifyNoMoreInteractions(mockResultSender);
+  }
+
+  @Test
+  @SuppressWarnings("deprecation")
+  public void lastResultFlushesAndCloses() {
+    final BufferedResultSender<Integer> resultSender = new BufferedResultSender<>(mockResultSender);
+
+    resultSender.lastResult(1);
+
+    verify(mockResultSender).lastResult(1);
+    verifyNoMoreInteractions(mockResultSender);
+  }
+
+  @Test
+  public void sendExceptionWithoutResultsSendsException() {
+    final Exception throwable = new Exception();
+
+    try (final BufferedResultSender<Integer> resultSender = buffered(mockResultSender)) {
+      resultSender.sendException(throwable);
+    }
+
+    verify(mockResultSender).sendException(throwable);
+    verifyNoMoreInteractions(mockResultSender);
+  }
+
+  @Test
+  public void sendExceptionWithoutResultFlushesAndSendsException() {
+    final Exception throwable = new Exception();
+
+    try (final BufferedResultSender<Integer> resultSender = buffered(mockResultSender)) {
+      resultSender.sendResult(1);
+      resultSender.sendException(throwable);
+    }
+
+    verify(mockResultSender).sendResult(1);
+    verify(mockResultSender).sendException(throwable);
+    verifyNoMoreInteractions(mockResultSender);
+  }
+
+  @Test
+  public void sendAllResults() {
+    try (final BufferedResultSender<Integer> resultSender = buffered(mockResultSender)) {
+      resultSender.sendAllResults(asList(1, 2, 3));
+    }
+
+    verify(mockResultSender).sendResult(1);
+    verify(mockResultSender).sendResult(2);
+    verify(mockResultSender).lastResult(3);
+    verifyNoMoreInteractions(mockResultSender);
+  }
+
+
+  @Test
+  public void sendAllResultsAfterCloseIsNoop() {
+    final BufferedResultSender<Integer> resultSender = new BufferedResultSender<>(mockResultSender);
+
+    resultSender.close();
+    resultSender.sendAllResults(asList(1, 2, 3));
+
+    verify(mockResultSender).lastResult(null);
+    verifyNoMoreInteractions(mockResultSender);
+  }
+}