You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by jb...@apache.org on 2019/08/23 07:51:29 UTC
[geode] branch wip/BufferedResultSender created (now 9b87df0)
This is an automated email from the ASF dual-hosted git repository.
jbarrett pushed a change to branch wip/BufferedResultSender
in repository https://gitbox.apache.org/repos/asf/geode.git.
at 9b87df0 BufferedResultSender to make function result sending feel better.
This branch includes the following new commits:
new a8b5451 BufferedResultSender to make function result sending feel better.
new 9b87df0 BufferedResultSender to make function result sending feel better.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[geode] 01/02: BufferedResultSender to make function result sending
feel better.
Posted by 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);
+ }
+}
[geode] 02/02: BufferedResultSender to make function result sending
feel better.
Posted by 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 9b87df065f051f377cc9c7bc1798c77687e4a7a8
Author: Jacob Barrett <jb...@pivotal.io>
AuthorDate: Fri Aug 23 00:50:56 2019 -0700
BufferedResultSender to make function result sending feel better.
---
.../java/org/apache/geode/cache/execute/BufferedResultSenderTest.java | 1 -
1 file changed, 1 deletion(-)
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
index 457787d..c166178 100644
--- 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
@@ -154,7 +154,6 @@ public class BufferedResultSenderTest {
verifyNoMoreInteractions(mockResultSender);
}
-
@Test
public void sendAllResultsAfterCloseIsNoop() {
final BufferedResultSender<Integer> resultSender = new BufferedResultSender<>(mockResultSender);