You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by bu...@apache.org on 2020/11/05 21:16:43 UTC
[geode] branch support/1.13 updated (6b8f5b0 -> 986334e)
This is an automated email from the ASF dual-hosted git repository.
burcham pushed a change to branch support/1.13
in repository https://gitbox.apache.org/repos/asf/geode.git.
from 6b8f5b0 Revert "GEODE-8496: Bump jetty from 9.4.21.v20190926 to 9.4.33.v20201020 (#5574)"
new ef74657 Revert "GEODE-8652: NioSslEngine.close() Bypasses Locks (#5666)"
new 4886d20 Revert "GEODE-8540: Create new DistributedBlackboard Rule (#5557)"
new 986334e Revert "GEODE-8136: Move UncheckedUtils to geode-common (#5123)"
The 3 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.
Summary of changes:
.../apache/geode/util/internal/UncheckedUtils.java | 34 --
.../geode/util/internal/UncheckedUtilsTest.java | 60 ----
.../cache/PartitionedRegionSingleHopDUnitTest.java | 14 +-
...istributedRegionFunctionExecutionDUnitTest.java | 59 +++-
...oningWithColocationAndPersistenceDUnitTest.java | 12 +-
.../tcp/ConnectionCloseSSLTLSDUnitTest.java | 238 -------------
.../org/apache/geode/internal/tcp/server.keystore | Bin 1256 -> 0 bytes
.../FunctionExecutionOnLonerRegressionTest.java | 27 +-
...LSocketHostNameVerificationIntegrationTest.java | 4 +-
.../internal/net/SSLSocketIntegrationTest.java | 57 ++-
.../apache/geode/codeAnalysis/excludedClasses.txt | 1 -
.../client/internal/ClientMetadataService.java | 4 +-
.../geode/internal/cache/GemFireCacheImpl.java | 49 ++-
.../cache/InternalCacheForClientAccess.java | 6 +-
.../apache/geode/internal/cache/LocalRegion.java | 7 +-
.../cache/execute/util/TypedFunctionService.java | 37 --
.../tier/sockets/CacheClientProxyFactory.java | 4 +-
.../AbortBackup.java => util/UncheckedUtils.java} | 19 +-
.../geode/internal/net/ByteBufferSharing.java | 55 ---
.../geode/internal/net/ByteBufferSharingImpl.java | 148 --------
.../geode/internal/net/ByteBufferSharingNoOp.java | 52 ---
.../org/apache/geode/internal/net/NioFilter.java | 69 ++--
.../apache/geode/internal/net/NioPlainEngine.java | 27 +-
.../apache/geode/internal/net/NioSslEngine.java | 353 +++++++++----------
.../org/apache/geode/internal/tcp/Connection.java | 34 +-
.../org/apache/geode/internal/tcp/MsgReader.java | 15 +-
.../internal/ClusterAlertMessagingTest.java | 6 +-
.../internal/net/ByteBufferSharingImplTest.java | 163 ---------
.../geode/internal/net/NioPlainEngineTest.java | 47 +--
.../geode/internal/net/NioSslEngineTest.java | 392 ++++++++++-----------
.../apache/geode/internal/tcp/TCPConduitTest.java | 4 +-
.../dunit/internal/DUnitBlackboardDUnitTest.java | 75 ++--
.../DistributedBlackboardDistributedTest.java | 297 ----------------
.../org/apache/geode/test/dunit/Blackboard.java | 79 -----
.../apache/geode/test/dunit/DUnitBlackboard.java | 55 ++-
.../test/dunit/internal/InternalBlackboard.java | 33 +-
.../dunit/internal/InternalBlackboardImpl.java | 59 +---
.../test/dunit/rules/DistributedBlackboard.java | 138 --------
38 files changed, 696 insertions(+), 2037 deletions(-)
delete mode 100644 geode-common/src/main/java/org/apache/geode/util/internal/UncheckedUtils.java
delete mode 100644 geode-common/src/test/java/org/apache/geode/util/internal/UncheckedUtilsTest.java
delete mode 100644 geode-core/src/distributedTest/java/org/apache/geode/internal/tcp/ConnectionCloseSSLTLSDUnitTest.java
delete mode 100644 geode-core/src/distributedTest/resources/org/apache/geode/internal/tcp/server.keystore
delete mode 100644 geode-core/src/main/java/org/apache/geode/internal/cache/execute/util/TypedFunctionService.java
copy geode-core/src/main/java/org/apache/geode/internal/cache/{backup/AbortBackup.java => util/UncheckedUtils.java} (70%)
delete mode 100644 geode-core/src/main/java/org/apache/geode/internal/net/ByteBufferSharing.java
delete mode 100644 geode-core/src/main/java/org/apache/geode/internal/net/ByteBufferSharingImpl.java
delete mode 100644 geode-core/src/main/java/org/apache/geode/internal/net/ByteBufferSharingNoOp.java
delete mode 100644 geode-core/src/test/java/org/apache/geode/internal/net/ByteBufferSharingImplTest.java
delete mode 100644 geode-dunit/src/distributedTest/java/org/apache/geode/test/dunit/rules/tests/DistributedBlackboardDistributedTest.java
delete mode 100644 geode-dunit/src/main/java/org/apache/geode/test/dunit/Blackboard.java
delete mode 100644 geode-dunit/src/main/java/org/apache/geode/test/dunit/rules/DistributedBlackboard.java
[geode] 02/03: Revert "GEODE-8540: Create new DistributedBlackboard
Rule (#5557)"
Posted by bu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
burcham pushed a commit to branch support/1.13
in repository https://gitbox.apache.org/repos/asf/geode.git
commit 4886d2055f9cd0792694d0edb61537429a037439
Author: Bill Burcham <bi...@gmail.com>
AuthorDate: Thu Nov 5 12:28:55 2020 -0800
Revert "GEODE-8540: Create new DistributedBlackboard Rule (#5557)"
This reverts commit cde469c6b6955a334e6bbf22accfc0735f0c70f4.
---
.../dunit/internal/DUnitBlackboardDUnitTest.java | 75 +++---
.../DistributedBlackboardDistributedTest.java | 297 ---------------------
.../org/apache/geode/test/dunit/Blackboard.java | 79 ------
.../apache/geode/test/dunit/DUnitBlackboard.java | 55 ++--
.../test/dunit/internal/InternalBlackboard.java | 33 +--
.../dunit/internal/InternalBlackboardImpl.java | 59 ++--
.../test/dunit/rules/DistributedBlackboard.java | 138 ----------
7 files changed, 101 insertions(+), 635 deletions(-)
diff --git a/geode-dunit/src/distributedTest/java/org/apache/geode/test/dunit/internal/DUnitBlackboardDUnitTest.java b/geode-dunit/src/distributedTest/java/org/apache/geode/test/dunit/internal/DUnitBlackboardDUnitTest.java
index 5e151d7..ae78247 100755
--- a/geode-dunit/src/distributedTest/java/org/apache/geode/test/dunit/internal/DUnitBlackboardDUnitTest.java
+++ b/geode-dunit/src/distributedTest/java/org/apache/geode/test/dunit/internal/DUnitBlackboardDUnitTest.java
@@ -14,70 +14,83 @@
*/
package org.apache.geode.test.dunit.internal;
-import static java.util.concurrent.TimeUnit.SECONDS;
-import static org.apache.geode.test.dunit.VM.getVM;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.catchThrowable;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.junit.Test;
+import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.VM;
-@SuppressWarnings("serial")
-public class DUnitBlackboardDUnitTest extends JUnit4DistributedTestCase {
+public class DUnitBlackboardDUnitTest extends JUnit4DistributedTestCase {
@Test
- public void canPassDataBetweenVMs() {
+ public void canPassDataBetweenVMs() throws Exception {
final String MBOX = "myMailbox";
- VM vm0 = getVM(0);
- VM vm1 = getVM(1);
+ VM vm0 = Host.getHost(0).getVM(0);
+ VM vm1 = Host.getHost(0).getVM(1);
vm0.invoke("put data in mailbox", () -> getBlackboard().setMailbox(MBOX, "testing"));
- String result = vm1.invoke("get data from mailbox", () -> getBlackboard().getMailbox(MBOX));
+ String result = (String) vm1.invoke("get data from mailbox", () -> {
+ return getBlackboard().getMailbox(MBOX);
+ });
- assertThat(result).isEqualTo("testing");
+ assertEquals("testing", result);
}
@Test
- public void canSignalAnotherVM() {
+ public void canSignalAnotherVM() throws Exception {
final String GATE = "myGate";
- VM vm0 = getVM(0);
- VM vm1 = getVM(1);
+ VM vm0 = Host.getHost(0).getVM(0);
+ VM vm1 = Host.getHost(0).getVM(1);
vm1.invoke("wait on gate not yet signalled", () -> {
- assertThat(getBlackboard().isGateSignaled(GATE)).isFalse();
-
- Throwable thrown = catchThrowable(() -> {
- getBlackboard().waitForGate(GATE, 1, SECONDS);
- });
-
- assertThat(thrown).isInstanceOf(TimeoutException.class);
+ assertFalse(getBlackboard().isGateSignaled(GATE));
+ try {
+ getBlackboard().waitForGate(GATE, 1, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ // expected
+ return;
+ } catch (InterruptedException e) {
+ fail("unexpected interrupt");
+ }
+ fail("unexpected success");
});
vm0.invoke("signal gate", () -> getBlackboard().signalGate(GATE));
- vm1.invoke("wait on gate not yet signalled",
- () -> getBlackboard().waitForGate(GATE, 1, SECONDS));
+ vm1.invoke("wait on gate not yet signalled", () -> {
+ try {
+ getBlackboard().waitForGate(GATE, 1, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ fail("unexpected timeout");
+ } catch (InterruptedException e) {
+ fail("unexpected interrupt");
+ }
+ // success expected
+ });
}
@Test
- public void initBlackboardClearsEverything() {
+ public void initBlackboardClearsEverything() throws Exception {
for (int i = 0; i < 100; i++) {
getBlackboard().setMailbox("MBOX" + i, "value" + i);
- assertThat((Object) getBlackboard().getMailbox("MBOX" + i)).isEqualTo("value" + i);
-
+ assertEquals("value" + i, getBlackboard().getMailbox("MBOX" + i));
getBlackboard().signalGate("GATE" + i);
- assertThat(getBlackboard().isGateSignaled("GATE" + i)).isTrue();
+ assertTrue(getBlackboard().isGateSignaled("GATE" + i));
}
-
- getVM(1).invoke("clear blackboard", () -> getBlackboard().initBlackboard());
+ Host.getHost(0).getVM(1).invoke("clear blackboard", () -> getBlackboard().initBlackboard());
for (int i = 0; i < 100; i++) {
- assertThat((Object) getBlackboard().getMailbox("MBOX" + i)).isNull();
- assertThat(getBlackboard().isGateSignaled("GATE" + i)).isFalse();
+ assertNull(getBlackboard().getMailbox("MBOX" + i));
+ assertFalse(getBlackboard().isGateSignaled("GATE" + i));
}
}
}
diff --git a/geode-dunit/src/distributedTest/java/org/apache/geode/test/dunit/rules/tests/DistributedBlackboardDistributedTest.java b/geode-dunit/src/distributedTest/java/org/apache/geode/test/dunit/rules/tests/DistributedBlackboardDistributedTest.java
deleted file mode 100644
index ea3ed2e..0000000
--- a/geode-dunit/src/distributedTest/java/org/apache/geode/test/dunit/rules/tests/DistributedBlackboardDistributedTest.java
+++ /dev/null
@@ -1,297 +0,0 @@
-/*
- * 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.test.dunit.rules.tests;
-
-import static java.util.Arrays.asList;
-import static java.util.concurrent.TimeUnit.SECONDS;
-import static org.apache.geode.test.dunit.VM.getController;
-import static org.apache.geode.test.dunit.VM.getVM;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.catchThrowable;
-
-import java.io.Serializable;
-import java.util.concurrent.TimeoutException;
-
-import org.junit.Rule;
-import org.junit.Test;
-
-import org.apache.geode.test.dunit.VM;
-import org.apache.geode.test.dunit.rules.DistributedBlackboard;
-import org.apache.geode.test.dunit.rules.DistributedRule;
-import org.apache.geode.test.junit.rules.serializable.SerializableTestName;
-
-@SuppressWarnings({"serial", "CodeBlock2Expr"})
-public class DistributedBlackboardDistributedTest implements Serializable {
-
- @Rule
- public DistributedRule distributedRule = new DistributedRule();
- @Rule
- public DistributedBlackboard blackboard = new DistributedBlackboard();
- @Rule
- public SerializableTestName testName = new SerializableTestName();
-
- @Test
- public void canPassDataBetweenVMs() {
- VM vm0 = getVM(0);
- VM vm1 = getVM(1);
-
- vm0.invoke("put data in mailbox", () -> blackboard.setMailbox(mailbox(), value()));
-
- String result = vm1.invoke("get data from mailbox", () -> blackboard.getMailbox(mailbox()));
-
- assertThat(result).isEqualTo(value());
- }
-
- @Test
- public void canSignalAnotherVM() {
- VM vm0 = getVM(0);
- VM vm1 = getVM(1);
-
- vm1.invoke("wait on gate not yet signalled", () -> {
- assertThat(blackboard.isGateSignaled(gate())).isFalse();
-
- Throwable thrown = catchThrowable(() -> {
- blackboard.waitForGate(gate(), 1, SECONDS);
- });
-
- assertThat(thrown).isInstanceOf(TimeoutException.class);
- });
-
- vm0.invoke("signal gate", () -> blackboard.signalGate(gate()));
-
- vm1.invoke("wait on gate not yet signalled", () -> blackboard.waitForGate(gate(), 1, SECONDS));
- }
-
- @Test
- public void initBlackboardClearsEverything() {
- for (int i = 0; i < 100; i++) {
- blackboard.setMailbox(mailbox(i), value(i));
- assertThat((Object) blackboard.getMailbox(mailbox(i))).isEqualTo(value(i));
-
- blackboard.signalGate(gate(i));
- assertThat(blackboard.isGateSignaled(gate(i))).isTrue();
- }
-
- getVM(1).invoke("clear blackboard", () -> blackboard.initBlackboard());
-
- for (int i = 0; i < 100; i++) {
- assertThat((Object) blackboard.getMailbox(mailbox(i))).isNull();
- assertThat(blackboard.isGateSignaled(gate(i))).isFalse();
- }
- }
-
- @Test
- public void getMailbox_returnsValueFromSameVM() {
- getVM(0).invoke(() -> blackboard.setMailbox(mailbox(), value()));
-
- getVM(0).invoke(() -> {
- assertThat((Object) blackboard.getMailbox(mailbox())).isEqualTo(value());
- });
- }
-
- @Test
- public void getMailbox_returnsValueFromOtherVM() {
- getVM(0).invoke(() -> blackboard.setMailbox(mailbox(), value()));
-
- getVM(1).invoke(() -> {
- assertThat((Object) blackboard.getMailbox(mailbox())).isEqualTo(value());
- });
- }
-
- @Test
- public void setMailbox_overwrites_valueFromSameVM() {
- getVM(0).invoke(() -> blackboard.setMailbox(mailbox(), value(1)));
- getVM(0).invoke(() -> blackboard.setMailbox(mailbox(), value(2)));
-
- getVM(0).invoke(() -> {
- assertThat((Object) blackboard.getMailbox(mailbox())).isEqualTo(value(2));
- });
- }
-
- @Test
- public void setMailbox_overwrites_valueFromOtherVM() {
- getVM(0).invoke(() -> blackboard.setMailbox(mailbox(), value(1)));
- getVM(1).invoke(() -> blackboard.setMailbox(mailbox(), value(2)));
-
- getVM(2).invoke(() -> {
- assertThat((Object) blackboard.getMailbox(mailbox())).isEqualTo(value(2));
- });
- }
-
- @Test
- public void getMailbox_returnsValueFromSameVM_afterBouncingVM() {
- getVM(0).invoke(() -> blackboard.setMailbox(mailbox(), value()));
-
- getVM(0).bounceForcibly();
-
- getVM(0).invoke(() -> {
- assertThat((Object) blackboard.getMailbox(mailbox())).isEqualTo(value());
- });
- }
-
- @Test
- public void getMailbox_returnsValueFromOtherVM_afterBouncingVM() {
- getVM(0).invoke(() -> blackboard.setMailbox(mailbox(), value()));
-
- getVM(0).bounceForcibly();
-
- getVM(1).invoke(() -> {
- assertThat((Object) blackboard.getMailbox(mailbox())).isEqualTo(value());
- });
- }
-
- @Test
- public void setMailbox_overwrites_valueFromSameVM_afterBouncingVM() {
- getVM(0).invoke(() -> blackboard.setMailbox(mailbox(), value(1)));
- getVM(0).invoke(() -> blackboard.setMailbox(mailbox(), value(2)));
-
- getVM(0).bounceForcibly();
-
- getVM(0).invoke(() -> {
- assertThat((Object) blackboard.getMailbox(mailbox(1))).isEqualTo(value(2));
- });
- }
-
- @Test
- public void setMailbox_overwrites_valueFromOtherVM_afterBouncingFirstVM() {
- getVM(0).invoke(() -> blackboard.setMailbox(mailbox(), value(1)));
- getVM(1).invoke(() -> blackboard.setMailbox(mailbox(), value(2)));
-
- getVM(0).bounceForcibly();
-
- getVM(2).invoke(() -> {
- assertThat((Object) blackboard.getMailbox(mailbox())).isEqualTo(value(2));
- });
- }
-
- @Test
- public void setMailbox_overwrites_valueFromOtherVM_afterBouncingSecondVM() {
- getVM(0).invoke(() -> blackboard.setMailbox(mailbox(), value(1)));
- getVM(1).invoke(() -> blackboard.setMailbox(mailbox(), value(2)));
-
- getVM(1).bounceForcibly();
-
- getVM(2).invoke(() -> {
- assertThat((Object) blackboard.getMailbox(mailbox())).isEqualTo(value(2));
- });
- }
-
- @Test
- public void setMailbox_overwrites_valueFromOtherVM_afterBouncingBothVMs() {
- getVM(0).invoke(() -> blackboard.setMailbox(mailbox(), value(1)));
- getVM(1).invoke(() -> blackboard.setMailbox(mailbox(), value(2)));
-
- getVM(0).bounceForcibly();
- getVM(1).bounceForcibly();
-
- getVM(2).invoke(() -> {
- assertThat((Object) blackboard.getMailbox(mailbox())).isEqualTo(value(2));
- });
- }
-
- @Test
- public void getMailbox_returnsValueFromSameVM_afterBouncingEveryVM() {
- getVM(0).invoke(() -> blackboard.setMailbox(mailbox(), value()));
-
- getVM(0).bounceForcibly();
- getVM(1).bounceForcibly();
- getVM(2).bounceForcibly();
- getVM(3).bounceForcibly();
-
- getVM(0).invoke(() -> {
- assertThat((Object) blackboard.getMailbox(mailbox())).isEqualTo(value());
- });
- }
-
- @Test
- public void getMailbox_returnsValueFromOtherVM_afterBouncingEveryVM() {
- getVM(0).invoke(() -> blackboard.setMailbox(mailbox(), value()));
-
- getVM(0).bounceForcibly();
- getVM(1).bounceForcibly();
- getVM(2).bounceForcibly();
- getVM(3).bounceForcibly();
-
- getVM(1).invoke(() -> {
- assertThat((Object) blackboard.getMailbox(mailbox())).isEqualTo(value());
- });
- }
-
- @Test
- public void getMailbox_returnsValueFromControllerVM_afterBouncingEveryVM() {
- blackboard.setMailbox(mailbox(), value());
-
- getVM(0).bounceForcibly();
- getVM(1).bounceForcibly();
- getVM(2).bounceForcibly();
- getVM(3).bounceForcibly();
-
- getVM(3).invoke(() -> {
- assertThat((Object) blackboard.getMailbox(mailbox())).isEqualTo(value());
- });
- }
-
- @Test
- public void getMailbox_returnsValueInControllerVM_afterBouncingEveryVM() {
- blackboard.setMailbox(mailbox(), value());
-
- getVM(0).bounceForcibly();
- getVM(1).bounceForcibly();
- getVM(2).bounceForcibly();
- getVM(3).bounceForcibly();
-
- assertThat((Object) blackboard.getMailbox(mailbox())).isEqualTo(value());
- }
-
- @Test
- public void getMailbox_returnsValueInEveryVM() {
- blackboard.setMailbox(mailbox(), value());
-
- assertThat((Object) blackboard.getMailbox(mailbox())).isEqualTo(value());
- for (VM vm : asList(getController(), getVM(0), getVM(1), getVM(2), getVM(3))) {
- vm.invoke(() -> {
- assertThat((Object) blackboard.getMailbox(mailbox())).isEqualTo(value());
- });
- }
- }
-
- private String mailbox() {
- return value("mailbox", 1);
- }
-
- private String value() {
- return value("value", 1);
- }
-
- private String gate() {
- return value("gate", 1);
- }
-
- private String mailbox(int count) {
- return value("mailbox", count);
- }
-
- private String value(int count) {
- return value("value", count);
- }
-
- private String gate(int count) {
- return value("gate", count);
- }
-
- private String value(String prefix, int count) {
- return prefix + "-" + testName.getMethodName() + "-" + count;
- }
-}
diff --git a/geode-dunit/src/main/java/org/apache/geode/test/dunit/Blackboard.java b/geode-dunit/src/main/java/org/apache/geode/test/dunit/Blackboard.java
deleted file mode 100644
index 2b15ebd..0000000
--- a/geode-dunit/src/main/java/org/apache/geode/test/dunit/Blackboard.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * 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.test.dunit;
-
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-import org.apache.geode.test.awaitility.GeodeAwaitility;
-
-/**
- * Blackboard provides mailboxes and synchronization gateways for distributed tests.
- *
- * <p>
- * Tests may use the blackboard to pass objects and status between JVMs with mailboxes instead of
- * using static variables in classes. The caveat being that the objects will be serialized using
- * Java serialization.
- *
- * <p>
- * Gates may be used to synchronize operations between distributed test JVMs. Combined with
- * Awaitility these can be used to test for conditions being met, actions having happened, etc.
- *
- * <p>
- * Look for references to the given methods in your IDE for examples.
- */
-public interface Blackboard {
-
- /**
- * Resets the blackboard.
- */
- void initBlackboard();
-
- /**
- * Signals a boolean gate.
- */
- void signalGate(String gateName);
-
- /**
- * Waits at most {@link GeodeAwaitility#getTimeout()} for a gate to be signaled.
- */
- void waitForGate(String gateName) throws TimeoutException, InterruptedException;
-
- /**
- * Waits at most the specified timeout for a gate to be signaled.
- */
- void waitForGate(String gateName, long timeout, TimeUnit units)
- throws TimeoutException, InterruptedException;
-
- /**
- * Clears a gate.
- */
- void clearGate(String gateName);
-
- /**
- * Checks to see if a gate has been signaled.
- */
- boolean isGateSignaled(String gateName);
-
- /**
- * Puts an object into a mailbox slot. The object must be java-serializable.
- */
- <T> void setMailbox(String boxName, T value);
-
- /**
- * Retrieves an object from a mailbox slot.
- */
- <T> T getMailbox(String boxName);
-}
diff --git a/geode-dunit/src/main/java/org/apache/geode/test/dunit/DUnitBlackboard.java b/geode-dunit/src/main/java/org/apache/geode/test/dunit/DUnitBlackboard.java
index 1ff5a0a..d87b99d 100755
--- a/geode-dunit/src/main/java/org/apache/geode/test/dunit/DUnitBlackboard.java
+++ b/geode-dunit/src/main/java/org/apache/geode/test/dunit/DUnitBlackboard.java
@@ -14,9 +14,6 @@
*/
package org.apache.geode.test.dunit;
-import static java.util.concurrent.TimeUnit.MINUTES;
-import static org.apache.geode.test.awaitility.GeodeAwaitility.getTimeout;
-
import java.rmi.RemoteException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -25,7 +22,7 @@ import org.apache.geode.test.dunit.internal.InternalBlackboard;
import org.apache.geode.test.dunit.internal.InternalBlackboardImpl;
/**
- * DUnitBlackboard provides mailboxes and synchronization gateways for distributed tests.
+ * DUnitBlackboard provides mailboxes and synchronization gateways for distributed unit tests.
*
* <p>
* Tests may use the blackboard to pass objects and status between JVMs with mailboxes instead of
@@ -39,19 +36,17 @@ import org.apache.geode.test.dunit.internal.InternalBlackboardImpl;
* <p>
* Look for references to the given methods in your IDE for examples.
*/
-public class DUnitBlackboard implements Blackboard {
+public class DUnitBlackboard {
- private final InternalBlackboard blackboard;
+ private InternalBlackboard blackboard;
public DUnitBlackboard() {
- this(InternalBlackboardImpl.getInstance());
- }
-
- public DUnitBlackboard(InternalBlackboard blackboard) {
- this.blackboard = blackboard;
+ blackboard = InternalBlackboardImpl.getInstance();
}
- @Override
+ /**
+ * resets the blackboard
+ */
public void initBlackboard() {
try {
blackboard.initBlackboard();
@@ -60,8 +55,11 @@ public class DUnitBlackboard implements Blackboard {
}
}
- @Override
+ /**
+ * signals a boolean gate
+ */
public void signalGate(String gateName) {
+ // System.out.println(Thread.currentThread().getName()+": signaling gate " + gateName);
try {
blackboard.signalGate(gateName);
} catch (RemoteException e) {
@@ -69,15 +67,12 @@ public class DUnitBlackboard implements Blackboard {
}
}
- @Override
- public void waitForGate(String gateName)
- throws TimeoutException, InterruptedException {
- waitForGate(gateName, getTimeout().toMinutes(), MINUTES);
- }
-
- @Override
+ /**
+ * wait for a gate to be signaled
+ */
public void waitForGate(String gateName, long timeout, TimeUnit units)
throws TimeoutException, InterruptedException {
+ // System.out.println(Thread.currentThread().getName()+": waiting for gate " + gateName);
try {
blackboard.waitForGate(gateName, timeout, units);
} catch (RemoteException e) {
@@ -85,7 +80,9 @@ public class DUnitBlackboard implements Blackboard {
}
}
- @Override
+ /**
+ * clear a gate
+ */
public void clearGate(String gateName) {
try {
blackboard.clearGate(gateName);
@@ -94,7 +91,9 @@ public class DUnitBlackboard implements Blackboard {
}
}
- @Override
+ /**
+ * test to see if a gate has been signeled
+ */
public boolean isGateSignaled(String gateName) {
try {
return blackboard.isGateSignaled(gateName);
@@ -103,7 +102,9 @@ public class DUnitBlackboard implements Blackboard {
}
}
- @Override
+ /**
+ * put an object into a mailbox slot. The object must be java-serializable
+ */
public void setMailbox(String boxName, Object value) {
try {
blackboard.setMailbox(boxName, value);
@@ -112,7 +113,9 @@ public class DUnitBlackboard implements Blackboard {
}
}
- @Override
+ /**
+ * retrieve an object from a mailbox slot
+ */
public <T> T getMailbox(String boxName) {
try {
return blackboard.getMailbox(boxName);
@@ -120,8 +123,4 @@ public class DUnitBlackboard implements Blackboard {
throw new RuntimeException("remote call failed", e);
}
}
-
- public InternalBlackboard internal() {
- return blackboard;
- }
}
diff --git a/geode-dunit/src/main/java/org/apache/geode/test/dunit/internal/InternalBlackboard.java b/geode-dunit/src/main/java/org/apache/geode/test/dunit/internal/InternalBlackboard.java
index 222a301..24abf4f 100755
--- a/geode-dunit/src/main/java/org/apache/geode/test/dunit/internal/InternalBlackboard.java
+++ b/geode-dunit/src/main/java/org/apache/geode/test/dunit/internal/InternalBlackboard.java
@@ -17,70 +17,61 @@ package org.apache.geode.test.dunit.internal;
import java.io.Serializable;
import java.rmi.Remote;
import java.rmi.RemoteException;
-import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/**
- * InternalBlackboard provides mailboxes and synchronization gateways for distributed tests.
- *
+ * DUnitBlackboard provides mailboxes and synchronization gateways for distributed unit tests.
* <p>
* Tests may use the blackboard to pass objects and status between JVMs with mailboxes instead of
* using static variables in classes. The caveat being that the objects will be serialized using
* Java serialization.
- *
* <p>
* Gates may be used to synchronize operations between unit test JVMs. Combined with Awaitility
* these can be used to test for conditions being met, actions having happened, etc.
+ * <p>
+ * Look for references to the given methods in your IDE for examples.
*/
public interface InternalBlackboard extends Remote, Serializable {
-
/**
- * Resets the blackboard.
+ * resets the blackboard
*/
void initBlackboard() throws RemoteException;
/**
- * Signals a boolean gate.
+ * signals a boolean gate
*/
void signalGate(String gateName) throws RemoteException;
/**
- * Waits for a gate to be signaled.
+ * wait for a gate to be signaled
*/
void waitForGate(String gateName, long timeout, TimeUnit units)
throws RemoteException, TimeoutException, InterruptedException;
/**
- * Clears a gate.
+ * clears a gate
*/
void clearGate(String gateName) throws RemoteException;
/**
- * Checks to see if a gate has been signaled.
+ * test to see if a gate has been signeled
*/
boolean isGateSignaled(String gateName) throws RemoteException;
/**
- * Puts an object into a mailbox slot. The object must be java-serializable.
+ * put an object into a mailbox slot. The object must be java-serializable
*/
- <T> void setMailbox(String boxName, T value) throws RemoteException;
+ void setMailbox(String boxName, Object value) throws RemoteException;
/**
- * Retrieves an object from a mailbox slot.
+ * retrieve an object from a mailbox slot
*/
<T> T getMailbox(String boxName) throws RemoteException;
/**
- * Pings the blackboard to make sure it's there.
+ * ping the blackboard to make sure it's there
*/
void ping() throws RemoteException;
- Map<String, Boolean> gates() throws RemoteException;
-
- Map<String, Serializable> mailboxes() throws RemoteException;
-
- void putGates(Map<String, Boolean> gates) throws RemoteException;
-
- void putMailboxes(Map<String, Serializable> mailboxes) throws RemoteException;
}
diff --git a/geode-dunit/src/main/java/org/apache/geode/test/dunit/internal/InternalBlackboardImpl.java b/geode-dunit/src/main/java/org/apache/geode/test/dunit/internal/InternalBlackboardImpl.java
index e24a5a0..bbed22e 100755
--- a/geode-dunit/src/main/java/org/apache/geode/test/dunit/internal/InternalBlackboardImpl.java
+++ b/geode-dunit/src/main/java/org/apache/geode/test/dunit/internal/InternalBlackboardImpl.java
@@ -14,12 +14,6 @@
*/
package org.apache.geode.test.dunit.internal;
-import static java.util.Collections.unmodifiableMap;
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
-
-import java.io.Serializable;
-import java.net.MalformedURLException;
import java.rmi.AlreadyBoundException;
import java.rmi.Naming;
import java.rmi.NotBoundException;
@@ -30,24 +24,26 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+
public class InternalBlackboardImpl extends UnicastRemoteObject implements InternalBlackboard {
+ public static InternalBlackboard blackboard;
- private static InternalBlackboard blackboard;
+ private Map<String, Boolean> gates = new ConcurrentHashMap<>();
+
+ private Map<String, Object> mailboxes = new ConcurrentHashMap();
- private final Map<String, Boolean> gates = new ConcurrentHashMap<>();
- private final Map<String, Serializable> mailboxes = new ConcurrentHashMap<>();
/**
* Zero-arg constructor for remote method invocations.
*/
public InternalBlackboardImpl() throws RemoteException {
- // nothing
+ super();
}
/**
* Creates a singleton event listeners blackboard.
*/
- public static synchronized InternalBlackboard getInstance() {
+ public static InternalBlackboard getInstance() {
if (blackboard == null) {
try {
initialize();
@@ -60,12 +56,11 @@ public class InternalBlackboardImpl extends UnicastRemoteObject implements Inter
return blackboard;
}
- private static synchronized void initialize()
- throws AlreadyBoundException, MalformedURLException, RemoteException {
+ private static synchronized void initialize() throws Exception {
if (blackboard == null) {
System.out.println(
DUnitLauncher.RMI_PORT_PARAM + "=" + System.getProperty(DUnitLauncher.RMI_PORT_PARAM));
- int namingPort = Integer.getInteger(DUnitLauncher.RMI_PORT_PARAM);
+ int namingPort = Integer.getInteger(DUnitLauncher.RMI_PORT_PARAM).intValue();
String name = "//localhost:" + namingPort + "/" + "InternalBlackboard";
try {
blackboard = (InternalBlackboard) Naming.lookup(name);
@@ -79,8 +74,8 @@ public class InternalBlackboardImpl extends UnicastRemoteObject implements Inter
@Override
public void initBlackboard() throws RemoteException {
- gates.clear();
- mailboxes.clear();
+ this.gates.clear();
+ this.mailboxes.clear();
}
@Override
@@ -95,8 +90,8 @@ public class InternalBlackboardImpl extends UnicastRemoteObject implements Inter
@Override
public void waitForGate(final String gateName, final long timeout, final TimeUnit units)
- throws InterruptedException, RemoteException, TimeoutException {
- long giveupTime = System.currentTimeMillis() + MILLISECONDS.convert(timeout, units);
+ throws RemoteException, TimeoutException, InterruptedException {
+ long giveupTime = System.currentTimeMillis() + TimeUnit.MILLISECONDS.convert(timeout, units);
while (System.currentTimeMillis() < giveupTime) {
Boolean gate = gates.get(gateName);
if (gate != null && gate) {
@@ -110,17 +105,17 @@ public class InternalBlackboardImpl extends UnicastRemoteObject implements Inter
@Override
public boolean isGateSignaled(final String gateName) {
Boolean gate = gates.get(gateName);
- return gate != null && gate;
+ return (gate != null && gate);
}
@Override
- public <T> void setMailbox(String boxName, T value) {
- mailboxes.put(boxName, (Serializable) value);
+ public void setMailbox(String boxName, Object value) {
+ mailboxes.put(boxName, value);
}
@Override
- public <T> T getMailbox(String boxName) {
- return uncheckedCast(mailboxes.get(boxName));
+ public Object getMailbox(String boxName) {
+ return mailboxes.get(boxName);
}
@Override
@@ -128,23 +123,5 @@ public class InternalBlackboardImpl extends UnicastRemoteObject implements Inter
// no-op
}
- @Override
- public Map<String, Boolean> gates() {
- return unmodifiableMap(gates);
- }
- @Override
- public Map<String, Serializable> mailboxes() {
- return unmodifiableMap(mailboxes);
- }
-
- @Override
- public void putGates(Map<String, Boolean> gates) {
- this.gates.putAll(gates);
- }
-
- @Override
- public void putMailboxes(Map<String, Serializable> mailboxes) {
- this.mailboxes.putAll(mailboxes);
- }
}
diff --git a/geode-dunit/src/main/java/org/apache/geode/test/dunit/rules/DistributedBlackboard.java b/geode-dunit/src/main/java/org/apache/geode/test/dunit/rules/DistributedBlackboard.java
deleted file mode 100644
index 8161b6d..0000000
--- a/geode-dunit/src/main/java/org/apache/geode/test/dunit/rules/DistributedBlackboard.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * 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.test.dunit.rules;
-
-import java.io.Serializable;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicReference;
-
-import org.apache.geode.test.dunit.Blackboard;
-import org.apache.geode.test.dunit.DUnitBlackboard;
-import org.apache.geode.test.dunit.VM;
-import org.apache.geode.test.dunit.internal.InternalBlackboard;
-import org.apache.geode.test.dunit.internal.InternalBlackboardImpl;
-
-/**
- * DistributedBlackboard provides mailboxes and synchronization gateways for distributed tests.
- *
- * <p>
- * Tests may use the blackboard to pass objects and status between JVMs with mailboxes instead of
- * using static variables in classes. The caveat being that the objects will be serialized using
- * Java serialization.
- *
- * <p>
- * Gates may be used to synchronize operations between distributed test JVMs. Combined with
- * Awaitility these can be used to test for conditions being met, actions having happened, etc.
- *
- * <p>
- * Look for references to the given methods in your IDE for examples.
- */
-@SuppressWarnings({"serial", "unused"})
-public class DistributedBlackboard extends AbstractDistributedRule implements Blackboard {
-
- private static final AtomicReference<DUnitBlackboard> BLACKBOARD = new AtomicReference<>();
- private static final AtomicReference<InternalBlackboard> INTERNAL = new AtomicReference<>();
-
- private final Map<Integer, Map<String, Boolean>> keepGates = new ConcurrentHashMap<>();
- private final Map<Integer, Map<String, Serializable>> keepMailboxes = new ConcurrentHashMap<>();
-
- @Override
- protected void before() {
- invoker().invokeInEveryVMAndController(() -> invokeBefore());
- }
-
- @Override
- protected void after() throws Throwable {
- invoker().invokeInEveryVMAndController(() -> invokeAfter());
- }
-
- @Override
- protected void afterCreateVM(VM vm) {
- vm.invoke(() -> invokeBefore());
- }
-
- @Override
- protected void beforeBounceVM(VM vm) {
- keepGates.put(vm.getId(), vm.invoke(() -> INTERNAL.get().gates()));
- keepMailboxes.put(vm.getId(), vm.invoke(() -> INTERNAL.get().mailboxes()));
- }
-
- @Override
- protected void afterBounceVM(VM vm) {
- Map<String, Boolean> keepGatesForVM = keepGates.remove(vm.getId());
- Map<String, Serializable> keepMailboxesForVM = keepMailboxes.remove(vm.getId());
-
- vm.invoke(() -> {
- invokeBefore();
- INTERNAL.get().putGates(keepGatesForVM);
- INTERNAL.get().putMailboxes(keepMailboxesForVM);
- });
- }
-
- private void invokeBefore() {
- InternalBlackboard internalBlackboard = InternalBlackboardImpl.getInstance();
- INTERNAL.set(internalBlackboard);
- BLACKBOARD.set(new DUnitBlackboard(internalBlackboard));
- }
-
- private void invokeAfter() {
- BLACKBOARD.set(null);
- INTERNAL.set(null);
- }
-
- @Override
- public void initBlackboard() {
- BLACKBOARD.get().initBlackboard();
- }
-
- @Override
- public void signalGate(String gateName) {
- BLACKBOARD.get().signalGate(gateName);
- }
-
- @Override
- public void waitForGate(String gateName) throws TimeoutException, InterruptedException {
- BLACKBOARD.get().waitForGate(gateName);
- }
-
- @Override
- public void waitForGate(String gateName, long timeout, TimeUnit units)
- throws TimeoutException, InterruptedException {
- BLACKBOARD.get().waitForGate(gateName, timeout, units);
- }
-
- @Override
- public void clearGate(String gateName) {
- BLACKBOARD.get().clearGate(gateName);
- }
-
- @Override
- public boolean isGateSignaled(String gateName) {
- return BLACKBOARD.get().isGateSignaled(gateName);
- }
-
- @Override
- public <T> void setMailbox(String boxName, T value) {
- BLACKBOARD.get().setMailbox(boxName, value);
- }
-
- @Override
- public <T> T getMailbox(String boxName) {
- return BLACKBOARD.get().getMailbox(boxName);
- }
-}
[geode] 03/03: Revert "GEODE-8136: Move UncheckedUtils to
geode-common (#5123)"
Posted by bu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
burcham pushed a commit to branch support/1.13
in repository https://gitbox.apache.org/repos/asf/geode.git
commit 986334e9198a1756b839d0d13028f4a846ea29b5
Author: Bill Burcham <bi...@gmail.com>
AuthorDate: Thu Nov 5 12:28:55 2020 -0800
Revert "GEODE-8136: Move UncheckedUtils to geode-common (#5123)"
This reverts commit 10af7ea015ec85ef02b2e972c7a3dd3ec23bcb7f.
---
.../geode/util/internal/UncheckedUtilsTest.java | 60 ----------------------
.../cache/PartitionedRegionSingleHopDUnitTest.java | 14 ++---
...istributedRegionFunctionExecutionDUnitTest.java | 59 ++++++++++++++-------
...oningWithColocationAndPersistenceDUnitTest.java | 12 ++---
.../FunctionExecutionOnLonerRegressionTest.java | 27 ++++++++--
.../client/internal/ClientMetadataService.java | 4 +-
.../geode/internal/cache/GemFireCacheImpl.java | 49 ++++++++++++++----
.../cache/InternalCacheForClientAccess.java | 6 +--
.../apache/geode/internal/cache/LocalRegion.java | 7 +--
.../cache/execute/util/TypedFunctionService.java | 37 -------------
.../tier/sockets/CacheClientProxyFactory.java | 4 +-
.../geode/internal/cache/util}/UncheckedUtils.java | 19 +++----
.../internal/ClusterAlertMessagingTest.java | 6 +--
.../apache/geode/internal/tcp/TCPConduitTest.java | 4 +-
14 files changed, 134 insertions(+), 174 deletions(-)
diff --git a/geode-common/src/test/java/org/apache/geode/util/internal/UncheckedUtilsTest.java b/geode-common/src/test/java/org/apache/geode/util/internal/UncheckedUtilsTest.java
deleted file mode 100644
index 7c282b7..0000000
--- a/geode-common/src/test/java/org/apache/geode/util/internal/UncheckedUtilsTest.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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.util.internal;
-
-import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.catchThrowable;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import org.junit.Test;
-
-@SuppressWarnings("unchecked")
-public class UncheckedUtilsTest {
-
- @Test
- public void uncheckedCast_rawList_empty() {
- List rawList = new ArrayList();
-
- List<String> value = uncheckedCast(rawList);
-
- assertThat(value).isSameAs(rawList);
- }
-
- @Test
- public void uncheckedCast_rawList_nonEmpty() {
- List rawList = new ArrayList();
- rawList.add("1");
- rawList.add("2");
-
- List<String> value = uncheckedCast(rawList);
-
- assertThat(value).isSameAs(rawList);
- }
-
- @Test
- public void uncheckedCast_rawList_wrongTypes() {
- List rawList = new ArrayList();
- rawList.add(1);
- rawList.add(2);
- List<String> wrongType = uncheckedCast(rawList);
-
- Throwable thrown = catchThrowable(() -> wrongType.get(0));
-
- assertThat(thrown).isInstanceOf(ClassCastException.class);
- }
-}
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/PartitionedRegionSingleHopDUnitTest.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/PartitionedRegionSingleHopDUnitTest.java
index 43019bf..943c503 100755
--- a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/PartitionedRegionSingleHopDUnitTest.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/PartitionedRegionSingleHopDUnitTest.java
@@ -22,6 +22,7 @@ import static org.apache.geode.cache.RegionShortcut.PARTITION;
import static org.apache.geode.cache.RegionShortcut.PARTITION_PERSISTENT;
import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_CLUSTER_CONFIGURATION;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
+import static org.apache.geode.internal.cache.util.UncheckedUtils.cast;
import static org.apache.geode.internal.lang.SystemPropertyHelper.GEMFIRE_PREFIX;
import static org.apache.geode.management.ManagementService.getExistingManagementService;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
@@ -31,7 +32,6 @@ import static org.apache.geode.test.dunit.VM.getController;
import static org.apache.geode.test.dunit.VM.getVM;
import static org.apache.geode.test.dunit.VM.getVMId;
import static org.apache.geode.test.dunit.rules.DistributedRule.getDistributedSystemProperties;
-import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
@@ -76,6 +76,7 @@ import org.apache.geode.cache.client.internal.ClientPartitionAdvisor;
import org.apache.geode.cache.client.internal.InternalClientCache;
import org.apache.geode.cache.execute.FunctionAdapter;
import org.apache.geode.cache.execute.FunctionContext;
+import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.execute.RegionFunctionContext;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.LocatorLauncher;
@@ -83,7 +84,6 @@ import org.apache.geode.distributed.ServerLauncher;
import org.apache.geode.distributed.internal.ServerLocation;
import org.apache.geode.internal.cache.BucketAdvisor.ServerBucketProfile;
import org.apache.geode.internal.cache.execute.InternalFunctionInvocationTargetException;
-import org.apache.geode.internal.cache.execute.util.TypedFunctionService;
import org.apache.geode.management.ManagementService;
import org.apache.geode.management.membership.MembershipEvent;
import org.apache.geode.management.membership.UniversalMembershipListenerAdapter;
@@ -1336,22 +1336,22 @@ public class PartitionedRegionSingleHopDUnitTest implements Serializable {
}
private void executeFunctions(Region<Object, Object> region) {
- TypedFunctionService.onRegion(region)
+ cast(FunctionService.onRegion(region))
.withFilter(filter(0))
.execute(new PutFunction())
.getResult();
- TypedFunctionService.onRegion(region)
+ cast(FunctionService.onRegion(region))
.withFilter(filter(0, 1))
.execute(new PutFunction())
.getResult();
- TypedFunctionService.onRegion(region)
+ cast(FunctionService.onRegion(region))
.withFilter(filter(0, 1, 2, 3))
.execute(new PutFunction())
.getResult();
- TypedFunctionService.onRegion(region)
+ cast(FunctionService.onRegion(region))
.execute(new PutFunction())
.getResult();
}
@@ -1389,7 +1389,7 @@ public class PartitionedRegionSingleHopDUnitTest implements Serializable {
}
private InternalCache getInternalCache(ServerLauncher serverLauncher) {
- return uncheckedCast(serverLauncher.getCache());
+ return cast(serverLauncher.getCache());
}
private void waitForLocalBucketsCreation() {
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/execute/DistributedRegionFunctionExecutionDUnitTest.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/execute/DistributedRegionFunctionExecutionDUnitTest.java
index b308b20..b95a68b 100755
--- a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/execute/DistributedRegionFunctionExecutionDUnitTest.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/execute/DistributedRegionFunctionExecutionDUnitTest.java
@@ -22,13 +22,13 @@ import static org.apache.geode.distributed.ConfigurationProperties.NAME;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_AUTHENTICATOR;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_AUTH_INIT;
import static org.apache.geode.distributed.ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER;
+import static org.apache.geode.internal.cache.execute.DistributedRegionFunctionExecutionDUnitTest.UncheckedUtils.cast;
import static org.apache.geode.test.awaitility.GeodeAwaitility.getTimeout;
import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
import static org.apache.geode.test.dunit.VM.getController;
import static org.apache.geode.test.dunit.VM.getVM;
import static org.apache.geode.test.dunit.VM.toArray;
import static org.apache.geode.util.internal.GeodeGlossary.GEMFIRE_PREFIX;
-import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;
@@ -59,6 +59,7 @@ import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.client.PoolFactory;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.client.internal.InternalClientCache;
+import org.apache.geode.cache.execute.Execution;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.cache.execute.FunctionException;
@@ -69,7 +70,6 @@ import org.apache.geode.cache.execute.ResultCollector;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.internal.cache.InternalCache;
-import org.apache.geode.internal.cache.execute.util.TypedFunctionService;
import org.apache.geode.security.templates.DummyAuthenticator;
import org.apache.geode.security.templates.UserPasswordAuthInit;
import org.apache.geode.test.dunit.AsyncInvocation;
@@ -257,7 +257,7 @@ public class DistributedRegionFunctionExecutionDUnitTest implements Serializable
empty.invoke(() -> populateRegion(200));
AsyncInvocation executeFunctionInReplicate1 = replicate1.invokeAsync(() -> {
- ResultCollector<String, List<String>> resultCollector = TypedFunctionService
+ ResultCollector<String, List<String>> resultCollector = FunctionServiceCast
.<Void, String, List<String>>onRegion(getRegion())
.withFilter(filter)
.execute(LongRunningFunction.class.getSimpleName(), getTimeout().toMillis(),
@@ -302,7 +302,7 @@ public class DistributedRegionFunctionExecutionDUnitTest implements Serializable
replicate1.invoke(() -> {
Throwable thrown = catchThrowable(() -> {
- TypedFunctionService
+ FunctionServiceCast
.<Void, String, List<String>>onRegion(getRegion())
.withFilter(filter)
.execute(LongRunningFunction.class.getSimpleName(), 1000, MILLISECONDS);
@@ -950,7 +950,7 @@ public class DistributedRegionFunctionExecutionDUnitTest implements Serializable
}
client.invoke(() -> {
- ResultCollector<Boolean, List<Boolean>> resultCollector = TypedFunctionService
+ ResultCollector<Boolean, List<Boolean>> resultCollector = FunctionServiceCast
.<Boolean, Boolean, List<Boolean>>onRegion(getRegion())
.setArguments(true)
.execute(inlineFunction("Success", true));
@@ -1158,13 +1158,13 @@ public class DistributedRegionFunctionExecutionDUnitTest implements Serializable
}
private void executeNoResultFunction() {
- TypedFunctionService
+ FunctionServiceCast
.onRegion(getRegion())
.execute(new NoResultFunction());
}
private List<Boolean> executeDistributedRegionFunction() {
- return TypedFunctionService
+ return FunctionServiceCast
.<Boolean, Boolean, List<Boolean>>onRegion(getRegion())
.withFilter(filter)
.setArguments(false)
@@ -1173,7 +1173,7 @@ public class DistributedRegionFunctionExecutionDUnitTest implements Serializable
}
private void executeThrowsRuntimeExceptionFunction() {
- TypedFunctionService
+ FunctionServiceCast
.<Void, Void, Void>onRegion(getRegion())
.withFilter(filter)
.execute(new ThrowsRuntimeExceptionFunction());
@@ -1187,7 +1187,7 @@ public class DistributedRegionFunctionExecutionDUnitTest implements Serializable
filter.add("execKey-" + 100 + i);
}
- ResultCollector<Object, List<Object>> resultCollector = TypedFunctionService
+ ResultCollector<Object, List<Object>> resultCollector = FunctionServiceCast
.<Boolean, Object, List<Object>>onRegion(getRegion())
.withFilter(filter)
.setArguments(true)
@@ -1200,7 +1200,7 @@ public class DistributedRegionFunctionExecutionDUnitTest implements Serializable
.as("First element of " + resultCollector.getResult())
.isInstanceOf(CustomRuntimeException.class);
- resultCollector = TypedFunctionService
+ resultCollector = FunctionServiceCast
.<Set<String>, Object, List<Object>>onRegion(getRegion())
.withFilter(filter)
.setArguments(filter)
@@ -1220,7 +1220,7 @@ public class DistributedRegionFunctionExecutionDUnitTest implements Serializable
private void executeNoLastResultFunction() {
Throwable thrown = catchThrowable(() -> {
- TypedFunctionService
+ FunctionServiceCast
.onRegion(getRegion())
.withFilter(filter)
.execute(new NoLastResultFunction())
@@ -1234,7 +1234,7 @@ public class DistributedRegionFunctionExecutionDUnitTest implements Serializable
private void executeUnregisteredFunction() {
FunctionService.unregisterFunction(new DistributedRegionFunction().getId());
- TypedFunctionService
+ FunctionServiceCast
.<Void, Boolean, List<Boolean>>onRegion(getRegion())
.withFilter(filter)
.execute(new DistributedRegionFunction())
@@ -1242,7 +1242,7 @@ public class DistributedRegionFunctionExecutionDUnitTest implements Serializable
}
private void executeFunctionFunctionInvocationTargetException() {
- ResultCollector<Integer, List<Integer>> resultCollector = TypedFunctionService
+ ResultCollector<Integer, List<Integer>> resultCollector = FunctionServiceCast
.<Boolean, Integer, List<Integer>>onRegion(getRegion())
.setArguments(true)
.execute(ThrowsFunctionInvocationTargetExceptionFunction.class.getSimpleName());
@@ -1253,7 +1253,7 @@ public class DistributedRegionFunctionExecutionDUnitTest implements Serializable
private void executeFunctionFunctionInvocationTargetExceptionWithoutHA() {
Throwable thrown = catchThrowable(() -> {
- TypedFunctionService
+ FunctionServiceCast
.<Boolean, Integer, List<Integer>>onRegion(getRegion())
.setArguments(true)
.execute(ThrowsFunctionInvocationTargetExceptionFunction.class.getSimpleName())
@@ -1268,7 +1268,7 @@ public class DistributedRegionFunctionExecutionDUnitTest implements Serializable
}
private void executeFunctionFunctionInvocationTargetException_ClientServer() {
- ResultCollector<Integer, List<Integer>> resultCollector = TypedFunctionService
+ ResultCollector<Integer, List<Integer>> resultCollector = FunctionServiceCast
.<Boolean, Integer, List<Integer>>onRegion(getRegion())
.setArguments(true)
.execute(ThrowsFunctionInvocationTargetExceptionFunction.class.getSimpleName());
@@ -1279,7 +1279,7 @@ public class DistributedRegionFunctionExecutionDUnitTest implements Serializable
private void executeFunctionFunctionInvocationTargetException_ClientServer_WithoutHA() {
Throwable thrown = catchThrowable(() -> {
- TypedFunctionService
+ FunctionServiceCast
.<Boolean, Integer, List<Integer>>onRegion(getRegion())
.setArguments(true)
.execute(ThrowsFunctionInvocationTargetExceptionFunction.class.getSimpleName())
@@ -1294,7 +1294,7 @@ public class DistributedRegionFunctionExecutionDUnitTest implements Serializable
}
private static <K, V> Region<K, V> getRegion() {
- return uncheckedCast(REGION.get());
+ return cast(REGION.get());
}
private static void setRegion(Region<?, ?> region) {
@@ -1327,8 +1327,7 @@ public class DistributedRegionFunctionExecutionDUnitTest implements Serializable
@Override
public void execute(FunctionContext<Object> context) {
if (context.getArguments() instanceof Set) {
- Set<Integer> arguments =
- uncheckedCast(context.getArguments());
+ Set<Integer> arguments = cast(context.getArguments());
for (int i = 0; i < arguments.size(); i++) {
context.getResultSender().sendResult(i);
}
@@ -1558,4 +1557,26 @@ public class DistributedRegionFunctionExecutionDUnitTest implements Serializable
return -1;
}
}
+
+ @SuppressWarnings("unchecked")
+ static class FunctionServiceCast {
+
+ /**
+ * Provide unchecked cast of FunctionService.onRegion.
+ */
+ static <IN, OUT, AGG> Execution<IN, OUT, AGG> onRegion(Region<?, ?> region) {
+ return FunctionService.onRegion(region);
+ }
+ }
+
+ @SuppressWarnings({"unchecked", "unused"})
+ static class UncheckedUtils {
+
+ /**
+ * Provide unchecked cast of specified Object.
+ */
+ static <T> T cast(Object object) {
+ return (T) object;
+ }
+ }
}
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/partitioned/fixed/FixedPartitioningWithColocationAndPersistenceDUnitTest.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/partitioned/fixed/FixedPartitioningWithColocationAndPersistenceDUnitTest.java
index 99db19a..ea7c834 100644
--- a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/partitioned/fixed/FixedPartitioningWithColocationAndPersistenceDUnitTest.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/partitioned/fixed/FixedPartitioningWithColocationAndPersistenceDUnitTest.java
@@ -36,10 +36,10 @@ import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitionin
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Quarter.Q2;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Quarter.Q3;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Quarter.Q4;
+import static org.apache.geode.internal.cache.util.UncheckedUtils.cast;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.VM.getVM;
import static org.apache.geode.test.dunit.VM.getVMId;
-import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;
import static org.mockito.Mockito.mock;
@@ -1502,9 +1502,9 @@ public class FixedPartitioningWithColocationAndPersistenceDUnitTest implements S
PartitionedRegionDataStore dataStore = partitionedRegion.getDataStore();
if (dataStore != null) {
- return uncheckedCast(new LocalDataSet(partitionedRegion, dataStore.getAllLocalBucketIds()));
+ return cast(new LocalDataSet(partitionedRegion, dataStore.getAllLocalBucketIds()));
}
- return uncheckedCast(new LocalDataSet(partitionedRegion, emptySet()));
+ return cast(new LocalDataSet(partitionedRegion, emptySet()));
}
private void validateQuartersData() throws ParseException {
@@ -1555,20 +1555,20 @@ public class FixedPartitioningWithColocationAndPersistenceDUnitTest implements S
InternalDistributedMember idmForShipment = shipments.getBucketPrimary(i);
// take all the keys from the shipment for each bucket
- Set<CustomerId> customerKey = uncheckedCast(customers.getBucketKeys(i));
+ Set<CustomerId> customerKey = cast(customers.getBucketKeys(i));
assertThat(customerKey).isNotNull();
for (CustomerId customerId : customerKey) {
assertThat(customers.get(customerId)).isNotNull();
- Set<OrderId> orderKey = uncheckedCast(orders.getBucketKeys(i));
+ Set<OrderId> orderKey = cast(orders.getBucketKeys(i));
for (OrderId orderId : orderKey) {
assertThat(orders.get(orderId)).isNotNull();
if (orderId.getCustomerId().equals(customerId)) {
assertThat(idmForOrder).isEqualTo(idmForCustomer);
}
- Set<ShipmentId> shipmentKey = uncheckedCast(shipments.getBucketKeys(i));
+ Set<ShipmentId> shipmentKey = cast(shipments.getBucketKeys(i));
for (ShipmentId shipmentId : shipmentKey) {
assertThat(shipments.get(shipmentId)).isNotNull();
if (shipmentId.getOrderId().equals(orderId)) {
diff --git a/geode-core/src/integrationTest/java/org/apache/geode/internal/cache/execute/FunctionExecutionOnLonerRegressionTest.java b/geode-core/src/integrationTest/java/org/apache/geode/internal/cache/execute/FunctionExecutionOnLonerRegressionTest.java
index d0c4cbc..08b75cf 100644
--- a/geode-core/src/integrationTest/java/org/apache/geode/internal/cache/execute/FunctionExecutionOnLonerRegressionTest.java
+++ b/geode-core/src/integrationTest/java/org/apache/geode/internal/cache/execute/FunctionExecutionOnLonerRegressionTest.java
@@ -19,7 +19,7 @@ import static org.apache.geode.cache.RegionShortcut.REPLICATE;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER;
-import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
+import static org.apache.geode.internal.cache.execute.FunctionExecutionOnLonerRegressionTest.UncheckedUtils.cast;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Collection;
@@ -35,14 +35,15 @@ import org.junit.experimental.categories.Category;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.Region;
+import org.apache.geode.cache.execute.Execution;
import org.apache.geode.cache.execute.FunctionContext;
+import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.execute.RegionFunctionContext;
import org.apache.geode.cache.execute.ResultCollector;
import org.apache.geode.cache.partition.PartitionRegionHelper;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.LonerDistributionManager;
import org.apache.geode.internal.cache.InternalCache;
-import org.apache.geode.internal.cache.execute.util.TypedFunctionService;
import org.apache.geode.test.junit.categories.FunctionServiceTest;
/**
@@ -87,7 +88,7 @@ public class FunctionExecutionOnLonerRegressionTest {
populateRegion(region);
- ResultCollector<Collection<String>, Collection<String>> resultCollector = TypedFunctionService
+ ResultCollector<Collection<String>, Collection<String>> resultCollector = FunctionServiceCast
.<Void, Collection<String>, Collection<String>>onRegion(region)
.withFilter(keysForGet)
.execute(new TestFunction(DataSetSupplier.PARTITIONED));
@@ -104,7 +105,7 @@ public class FunctionExecutionOnLonerRegressionTest {
populateRegion(region);
- ResultCollector<Collection<String>, Collection<String>> resultCollector = TypedFunctionService
+ ResultCollector<Collection<String>, Collection<String>> resultCollector = FunctionServiceCast
.<Void, Collection<String>, Collection<String>>onRegion(region)
.withFilter(keysForGet)
.execute(new TestFunction(DataSetSupplier.REPLICATE));
@@ -166,7 +167,7 @@ public class FunctionExecutionOnLonerRegressionTest {
@Override
public void execute(FunctionContext<String> context) {
RegionFunctionContext regionFunctionContext = (RegionFunctionContext) context;
- Set<String> keys = uncheckedCast(regionFunctionContext.getFilter());
+ Set<String> keys = cast(regionFunctionContext.getFilter());
String lastKey = keys.iterator().next();
keys.remove(lastKey);
@@ -184,4 +185,20 @@ public class FunctionExecutionOnLonerRegressionTest {
return getClass().getName();
}
}
+
+ @SuppressWarnings({"unchecked", "WeakerAccess"})
+ private static class FunctionServiceCast {
+
+ static <IN, OUT, AGG> Execution<IN, OUT, AGG> onRegion(Region<?, ?> region) {
+ return FunctionService.onRegion(region);
+ }
+ }
+
+ @SuppressWarnings({"unchecked", "unused"})
+ static class UncheckedUtils {
+
+ static <T> T cast(Object object) {
+ return (T) object;
+ }
+ }
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ClientMetadataService.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ClientMetadataService.java
index 8b89bb8..0aaa9df 100755
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ClientMetadataService.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ClientMetadataService.java
@@ -14,7 +14,7 @@
*/
package org.apache.geode.cache.client.internal;
-import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
+import static org.apache.geode.internal.cache.util.UncheckedUtils.cast;
import java.util.ArrayList;
import java.util.Collection;
@@ -230,7 +230,7 @@ public class ClientMetadataService {
for (Map.Entry entry : serverToBuckets.entrySet()) {
ServerLocation server = (ServerLocation) entry.getKey();
- Set<Integer> buckets = uncheckedCast(entry.getValue());
+ Set<Integer> buckets = cast(entry.getValue());
for (Integer bucket : buckets) {
// use LinkedHashSet to maintain the order of keys
// the keys will be iterated several times
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
index 2cd1f3c..2f6b0b4 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
@@ -33,6 +33,10 @@ import static org.apache.geode.distributed.internal.ClusterDistributionManager.L
import static org.apache.geode.distributed.internal.DistributionConfig.DEFAULT_DURABLE_CLIENT_ID;
import static org.apache.geode.distributed.internal.InternalDistributedSystem.getAnyInstance;
import static org.apache.geode.internal.cache.ColocationHelper.getColocatedChildRegions;
+import static org.apache.geode.internal.cache.GemFireCacheImpl.UncheckedUtils.asDistributedMemberSet;
+import static org.apache.geode.internal.cache.GemFireCacheImpl.UncheckedUtils.createMapArray;
+import static org.apache.geode.internal.cache.GemFireCacheImpl.UncheckedUtils.uncheckedCast;
+import static org.apache.geode.internal.cache.GemFireCacheImpl.UncheckedUtils.uncheckedRegionAttributes;
import static org.apache.geode.internal.cache.LocalRegion.setThreadInitLevelRequirement;
import static org.apache.geode.internal.cache.PartitionedRegion.DISK_STORE_FLUSHED;
import static org.apache.geode.internal.cache.PartitionedRegion.OFFLINE_EQUAL_PERSISTED;
@@ -40,11 +44,11 @@ import static org.apache.geode.internal.cache.PartitionedRegion.PRIMARY_BUCKETS_
import static org.apache.geode.internal.cache.PartitionedRegionHelper.PARTITION_LOCK_SERVICE_NAME;
import static org.apache.geode.internal.cache.control.InternalResourceManager.ResourceType.HEAP_MEMORY;
import static org.apache.geode.internal.cache.control.InternalResourceManager.ResourceType.OFFHEAP_MEMORY;
+import static org.apache.geode.internal.cache.util.UncheckedUtils.cast;
import static org.apache.geode.internal.logging.CoreLoggingExecutors.newThreadPoolWithFixedFeed;
import static org.apache.geode.internal.tcp.ConnectionTable.threadWantsSharedResources;
import static org.apache.geode.logging.internal.executors.LoggingExecutors.newFixedThreadPool;
import static org.apache.geode.util.internal.GeodeGlossary.GEMFIRE_PREFIX;
-import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
@@ -1808,7 +1812,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
&& partitionedRegion.getDataPolicy() == DataPolicy.PERSISTENT_PARTITION) {
int numBuckets = partitionedRegion.getTotalNumberOfBuckets();
Map<InternalDistributedMember, PersistentMemberID>[] bucketMaps =
- uncheckedCast(new Map[numBuckets]);
+ createMapArray(numBuckets);
PartitionedRegionDataStore dataStore = partitionedRegion.getDataStore();
// lock all the primary buckets
@@ -2662,18 +2666,18 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
@Override
public Set<DistributedMember> getAdminMembers() {
- return uncheckedCast(dm.getAdminMemberSet());
+ return asDistributedMemberSet(dm.getAdminMemberSet());
}
@Override
public Set<DistributedMember> getMembers(Region region) {
if (region instanceof DistributedRegion) {
DistributedRegion distributedRegion = (DistributedRegion) region;
- return uncheckedCast(distributedRegion.getDistributionAdvisor().adviseCacheOp());
+ return asDistributedMemberSet(distributedRegion.getDistributionAdvisor().adviseCacheOp());
}
if (region instanceof PartitionedRegion) {
PartitionedRegion partitionedRegion = (PartitionedRegion) region;
- return uncheckedCast(partitionedRegion.getRegionAdvisor().adviseAllPRNodes());
+ return asDistributedMemberSet(partitionedRegion.getRegionAdvisor().adviseAllPRNodes());
}
return emptySet();
}
@@ -3059,14 +3063,15 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
system.handleResourceEvent(ResourceEvent.REGION_CREATE, region);
}
- return uncheckedCast(region);
+ return cast(region);
}
@Override
public <K, V> RegionAttributes<K, V> invokeRegionBefore(InternalRegion parent, String name,
RegionAttributes<K, V> attrs, InternalRegionArguments internalRegionArgs) {
for (RegionListener listener : regionListeners) {
- attrs = uncheckedCast(listener.beforeCreate(parent, name, attrs, internalRegionArgs));
+ attrs =
+ uncheckedRegionAttributes(listener.beforeCreate(parent, name, attrs, internalRegionArgs));
}
return attrs;
}
@@ -3182,7 +3187,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
@Override
public <K, V> Region<K, V> getRegionByPath(String path) {
- return uncheckedCast(getInternalRegionByPath(path));
+ return cast(getInternalRegionByPath(path));
}
@Override
@@ -3239,7 +3244,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
stopper.checkCancelInProgress(null);
return null;
}
- return uncheckedCast(result);
+ return cast(result);
}
String[] pathParts = parsePath(path);
@@ -3263,7 +3268,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
logger.debug("GemFireCache.getRegion, calling getSubregion on rootRegion({}): {}",
pathParts[0], pathParts[1]);
}
- return uncheckedCast(rootRegion.getSubregion(pathParts[1], returnDestroyedRegion));
+ return cast(rootRegion.getSubregion(pathParts[1], returnDestroyedRegion));
}
@Override
@@ -4053,7 +4058,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
@Override
public <K, V> RegionAttributes<K, V> getRegionAttributes(String id) {
- return uncheckedCast(namedRegionAttributes.get(id));
+ return GemFireCacheImpl.UncheckedUtils.<K, V>uncheckedCast(namedRegionAttributes).get(id);
}
@Override
@@ -5171,6 +5176,28 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
}
}
+ @SuppressWarnings("unchecked")
+ static class UncheckedUtils {
+
+ static Map<InternalDistributedMember, PersistentMemberID>[] createMapArray(int size) {
+ return new Map[size];
+ }
+
+ static Set<DistributedMember> asDistributedMemberSet(
+ Set<InternalDistributedMember> internalDistributedMembers) {
+ return (Set) internalDistributedMembers;
+ }
+
+ static <K, V> RegionAttributes<K, V> uncheckedRegionAttributes(RegionAttributes region) {
+ return region;
+ }
+
+ static <K, V> Map<String, RegionAttributes<K, V>> uncheckedCast(
+ Map<String, RegionAttributes<?, ?>> namedRegionAttributes) {
+ return (Map) namedRegionAttributes;
+ }
+ }
+
@FunctionalInterface
@VisibleForTesting
interface TXManagerImplFactory {
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCacheForClientAccess.java b/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCacheForClientAccess.java
index dbbb98b..10635fe 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCacheForClientAccess.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCacheForClientAccess.java
@@ -16,7 +16,7 @@
*/
package org.apache.geode.internal.cache;
-import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
+import static org.apache.geode.internal.cache.util.UncheckedUtils.cast;
import java.io.File;
import java.io.IOException;
@@ -162,7 +162,7 @@ public class InternalCacheForClientAccess implements InternalCache {
public <K, V> Region<K, V> getRegion(String path, boolean returnDestroyedRegion) {
Region result = delegate.getRegion(path, returnDestroyedRegion);
checkForInternalRegion(result);
- return uncheckedCast(result);
+ return cast(result);
}
@Override
@@ -176,7 +176,7 @@ public class InternalCacheForClientAccess implements InternalCache {
public <K, V> Region<K, V> getRegionByPath(String path) {
InternalRegion result = delegate.getInternalRegionByPath(path);
checkForInternalRegion(result);
- return uncheckedCast(result);
+ return cast(result);
}
@Override
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
index 098115b..849187b 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
@@ -17,9 +17,9 @@ package org.apache.geode.internal.cache;
import static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.AFTER_INITIAL_IMAGE;
import static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.ANY_INIT;
import static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.BEFORE_INITIAL_IMAGE;
+import static org.apache.geode.internal.cache.util.UncheckedUtils.cast;
import static org.apache.geode.internal.lang.SystemUtils.getLineSeparator;
import static org.apache.geode.internal.offheap.annotations.OffHeapIdentifier.ENTRY_EVENT_NEW_VALUE;
-import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
import java.io.DataInputStream;
import java.io.DataOutputStream;
@@ -8943,10 +8943,7 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
txState.getRealDeal(null, this);
}
try {
- proxyResult = getServerProxy().putAll(
- uncheckedCast(map),
- eventId,
- !event.isGenerateCallbacks(),
+ proxyResult = getServerProxy().putAll(cast(map), eventId, !event.isGenerateCallbacks(),
event.getCallbackArgument());
if (isDebugEnabled) {
logger.debug("PutAll received response from server: {}", proxyResult);
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/execute/util/TypedFunctionService.java b/geode-core/src/main/java/org/apache/geode/internal/cache/execute/util/TypedFunctionService.java
deleted file mode 100644
index 3f21439..0000000
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/execute/util/TypedFunctionService.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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.internal.cache.execute.util;
-
-import org.apache.geode.cache.Region;
-import org.apache.geode.cache.execute.Execution;
-import org.apache.geode.cache.execute.FunctionService;
-
-/**
- * Utilities for casting and working around raw types in the {@link FunctionService} API.
- */
-@SuppressWarnings({"unchecked", "unused"})
-public class TypedFunctionService {
-
- protected TypedFunctionService() {
- // do not instantiate
- }
-
- /**
- * Adds parameterized type support to {@link FunctionService#onRegion(Region)}.
- */
- public static <IN, OUT, AGG> Execution<IN, OUT, AGG> onRegion(Region<?, ?> region) {
- return FunctionService.onRegion(region);
- }
-}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientProxyFactory.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientProxyFactory.java
index ae26aa7..a3def49 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientProxyFactory.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientProxyFactory.java
@@ -14,7 +14,7 @@
*/
package org.apache.geode.internal.cache.tier.sockets;
-import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
+import static org.apache.geode.internal.cache.util.UncheckedUtils.cast;
import java.lang.reflect.InvocationTargetException;
import java.net.Socket;
@@ -55,7 +55,7 @@ public class CacheClientProxyFactory {
}
try {
Class<InternalCacheClientProxyFactory> proxyClass =
- uncheckedCast(ClassPathLoader.getLatest().forName(proxyClassName));
+ cast(ClassPathLoader.getLatest().forName(proxyClassName));
return proxyClass.getConstructor().newInstance();
} catch (ClassNotFoundException | NoSuchMethodException | InstantiationException
| IllegalAccessException | InvocationTargetException e) {
diff --git a/geode-common/src/main/java/org/apache/geode/util/internal/UncheckedUtils.java b/geode-core/src/main/java/org/apache/geode/internal/cache/util/UncheckedUtils.java
similarity index 68%
rename from geode-common/src/main/java/org/apache/geode/util/internal/UncheckedUtils.java
rename to geode-core/src/main/java/org/apache/geode/internal/cache/util/UncheckedUtils.java
index 61dbd8d..c03e990 100644
--- a/geode-common/src/main/java/org/apache/geode/util/internal/UncheckedUtils.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/util/UncheckedUtils.java
@@ -12,23 +12,18 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-package org.apache.geode.util.internal;
+package org.apache.geode.internal.cache.util;
+
+import org.apache.geode.cache.execute.Execution;
-/**
- * Utilities for casting and working with unchecked raw types.
- */
@SuppressWarnings({"unchecked", "unused"})
public class UncheckedUtils {
- protected UncheckedUtils() {
- // do not instantiate
+ public static <T> T cast(Object object) {
+ return (T) object;
}
- /**
- * Casts an instance of a raw type to a parameterized type. Preference should be given to
- * converting all code from using raw types to using parameterized types when possible.
- */
- public static <T> T uncheckedCast(Object object) {
- return (T) object;
+ public static <IN, OUT, AGG> Execution<IN, OUT, AGG> cast(Execution execution) {
+ return execution;
}
}
diff --git a/geode-core/src/test/java/org/apache/geode/alerting/internal/ClusterAlertMessagingTest.java b/geode-core/src/test/java/org/apache/geode/alerting/internal/ClusterAlertMessagingTest.java
index eadf54b..f7bc8d8 100644
--- a/geode-core/src/test/java/org/apache/geode/alerting/internal/ClusterAlertMessagingTest.java
+++ b/geode-core/src/test/java/org/apache/geode/alerting/internal/ClusterAlertMessagingTest.java
@@ -14,7 +14,7 @@
*/
package org.apache.geode.alerting.internal;
-import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
+import static org.apache.geode.internal.cache.util.UncheckedUtils.cast;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;
import static org.mockito.ArgumentMatchers.anyLong;
@@ -141,7 +141,7 @@ public class ClusterAlertMessagingTest {
public void sendAlertLogsWarning_ifAlertingIOExceptionIsCaught() {
ExecutorService executor = currentThreadExecutorService();
ClusterDistributionManager distributionManager = mock(ClusterDistributionManager.class);
- Consumer<AlertingIOException> alertingIOExceptionLogger = uncheckedCast(mock(Consumer.class));
+ Consumer<AlertingIOException> alertingIOExceptionLogger = cast(mock(Consumer.class));
ClusterAlertMessaging clusterAlertMessaging =
spyClusterAlertMessaging(distributionManager, executor, alertingIOExceptionLogger);
doThrow(new AlertingIOException(new IOException("Cannot form connection to alert listener")))
@@ -162,7 +162,7 @@ public class ClusterAlertMessagingTest {
public void sendAlertLogsWarningOnce_ifAlertingIOExceptionIsCaught() {
ExecutorService executor = currentThreadExecutorService();
ClusterDistributionManager distributionManager = mock(ClusterDistributionManager.class);
- Consumer<AlertingIOException> alertingIOExceptionLogger = uncheckedCast(mock(Consumer.class));
+ Consumer<AlertingIOException> alertingIOExceptionLogger = cast(mock(Consumer.class));
ClusterAlertMessaging clusterAlertMessaging =
spyClusterAlertMessaging(distributionManager, executor, alertingIOExceptionLogger);
doThrow(new AlertingIOException(new IOException("Cannot form connection to alert listener")))
diff --git a/geode-core/src/test/java/org/apache/geode/internal/tcp/TCPConduitTest.java b/geode-core/src/test/java/org/apache/geode/internal/tcp/TCPConduitTest.java
index c0bf0a5..0c30ce2 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/tcp/TCPConduitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/tcp/TCPConduitTest.java
@@ -16,7 +16,7 @@
*/
package org.apache.geode.internal.tcp;
-import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
+import static org.apache.geode.internal.cache.util.UncheckedUtils.cast;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;
import static org.mockito.Mockito.anyBoolean;
@@ -62,7 +62,7 @@ public class TCPConduitTest {
@Before
public void setUp() throws Exception {
- membership = uncheckedCast(mock(Membership.class));
+ membership = cast(mock(Membership.class));
directChannel = mock(DirectChannel.class);
connectionTable = mock(ConnectionTable.class);
socketCreator = new SocketCreator(new SSLConfig.Builder().build());
[geode] 01/03: Revert "GEODE-8652: NioSslEngine.close() Bypasses
Locks (#5666)"
Posted by bu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
burcham pushed a commit to branch support/1.13
in repository https://gitbox.apache.org/repos/asf/geode.git
commit ef74657254c2b2707a31b43af52af1734b71e961
Author: Bill Burcham <bi...@gmail.com>
AuthorDate: Thu Nov 5 12:28:55 2020 -0800
Revert "GEODE-8652: NioSslEngine.close() Bypasses Locks (#5666)"
This reverts commit b2af727ce23fd155f3665e3db2ecee6e8f80fba7.
---
.../tcp/ConnectionCloseSSLTLSDUnitTest.java | 238 -------------
.../org/apache/geode/internal/tcp/server.keystore | Bin 1256 -> 0 bytes
...LSocketHostNameVerificationIntegrationTest.java | 4 +-
.../internal/net/SSLSocketIntegrationTest.java | 57 ++-
.../apache/geode/codeAnalysis/excludedClasses.txt | 1 -
.../geode/internal/net/ByteBufferSharing.java | 55 ---
.../geode/internal/net/ByteBufferSharingImpl.java | 148 --------
.../geode/internal/net/ByteBufferSharingNoOp.java | 52 ---
.../org/apache/geode/internal/net/NioFilter.java | 69 ++--
.../apache/geode/internal/net/NioPlainEngine.java | 27 +-
.../apache/geode/internal/net/NioSslEngine.java | 353 +++++++++----------
.../org/apache/geode/internal/tcp/Connection.java | 34 +-
.../org/apache/geode/internal/tcp/MsgReader.java | 15 +-
.../internal/net/ByteBufferSharingImplTest.java | 163 ---------
.../geode/internal/net/NioPlainEngineTest.java | 47 +--
.../geode/internal/net/NioSslEngineTest.java | 392 ++++++++++-----------
16 files changed, 460 insertions(+), 1195 deletions(-)
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/tcp/ConnectionCloseSSLTLSDUnitTest.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/tcp/ConnectionCloseSSLTLSDUnitTest.java
deleted file mode 100644
index 77fe9bf..0000000
--- a/geode-core/src/distributedTest/java/org/apache/geode/internal/tcp/ConnectionCloseSSLTLSDUnitTest.java
+++ /dev/null
@@ -1,238 +0,0 @@
-/*
- * 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.internal.tcp;
-
-import static org.apache.geode.distributed.ConfigurationProperties.CONSERVE_SOCKETS;
-import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_CLUSTER_CONFIGURATION;
-import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
-import static org.apache.geode.distributed.ConfigurationProperties.NAME;
-import static org.apache.geode.distributed.ConfigurationProperties.SOCKET_BUFFER_SIZE;
-import static org.apache.geode.distributed.ConfigurationProperties.SOCKET_LEASE_TIME;
-import static org.apache.geode.distributed.ConfigurationProperties.SSL_ENABLED_COMPONENTS;
-import static org.apache.geode.distributed.ConfigurationProperties.SSL_KEYSTORE;
-import static org.apache.geode.distributed.ConfigurationProperties.SSL_KEYSTORE_PASSWORD;
-import static org.apache.geode.distributed.ConfigurationProperties.SSL_PROTOCOLS;
-import static org.apache.geode.distributed.ConfigurationProperties.SSL_REQUIRE_AUTHENTICATION;
-import static org.apache.geode.distributed.ConfigurationProperties.SSL_TRUSTSTORE;
-import static org.apache.geode.distributed.ConfigurationProperties.SSL_TRUSTSTORE_PASSWORD;
-import static org.apache.geode.distributed.ConfigurationProperties.USE_CLUSTER_CONFIGURATION;
-import static org.apache.geode.test.dunit.VM.getVM;
-import static org.apache.geode.test.util.ResourceUtils.createTempFileFromResource;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.assertj.core.api.Fail.fail;
-
-import java.io.File;
-import java.io.Serializable;
-import java.nio.ByteBuffer;
-import java.util.Properties;
-import java.util.concurrent.TimeoutException;
-
-import org.apache.logging.log4j.Logger;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-
-import org.apache.geode.cache.Cache;
-import org.apache.geode.cache.CacheFactory;
-import org.apache.geode.cache.Region;
-import org.apache.geode.cache.RegionShortcut;
-import org.apache.geode.distributed.DistributedSystemDisconnectedException;
-import org.apache.geode.distributed.Locator;
-import org.apache.geode.distributed.internal.ClusterDistributionManager;
-import org.apache.geode.distributed.internal.DistributionMessage;
-import org.apache.geode.distributed.internal.DistributionMessageObserver;
-import org.apache.geode.distributed.internal.membership.gms.membership.GMSJoinLeave;
-import org.apache.geode.internal.cache.UpdateOperation.UpdateMessage;
-import org.apache.geode.logging.internal.log4j.api.LogService;
-import org.apache.geode.test.dunit.AsyncInvocation;
-import org.apache.geode.test.dunit.VM;
-import org.apache.geode.test.dunit.rules.DistributedBlackboard;
-import org.apache.geode.test.dunit.rules.DistributedRestoreSystemProperties;
-import org.apache.geode.test.dunit.rules.DistributedRule;
-
-/**
- * It would be nice if this test didn't need to use the cache since the test's purpose is to test
- * that the {@link Connection} class can be closed while readers and writers hold locks on its
- * internal TLS {@link ByteBuffer}s
- *
- * But this test does use the cache (region) because it enabled us to use existing cache messaging
- * and to use the DistributionMessageObserver (observer) hooks.
- *
- * see also ClusterCommunicationsDUnitTest
- */
-public class ConnectionCloseSSLTLSDUnitTest implements Serializable {
-
- private static final int SMALL_BUFFER_SIZE = 8000;
- private static final String UPDATE_ENTERED_GATE = "connectionCloseDUnitTest.regionUpdateEntered";
- private static final String SUSPEND_UPDATE_GATE = "connectionCloseDUnitTest.suspendRegionUpdate";
- private static final String regionName = "connectionCloseDUnitTestRegion";
- private static final Logger logger = LogService.getLogger();
-
- private static Cache cache;
-
- @Rule
- public DistributedRule distributedRule =
- DistributedRule.builder().withVMCount(3).build();
-
- @Rule
- public DistributedBlackboard blackboard = new DistributedBlackboard();
-
- @Rule
- public DistributedRestoreSystemProperties restoreSystemProperties =
- new DistributedRestoreSystemProperties();
-
- private VM locator;
- private VM sender;
- private VM receiver;
-
- @Before
- public void before() {
- locator = getVM(0);
- sender = getVM(1);
- receiver = getVM(2);
- }
-
- @After
- public void after() {
- receiver.invoke(() -> {
- DistributionMessageObserver.setInstance(null);
- });
- }
-
- @Test
- public void connectionWithHungReaderIsCloseableAndUnhangsReader()
- throws InterruptedException, TimeoutException {
-
- blackboard.clearGate(UPDATE_ENTERED_GATE);
- blackboard.clearGate(SUSPEND_UPDATE_GATE);
-
- final int locatorPort = createLocator(locator);
- createCacheAndRegion(sender, locatorPort);
- createCacheAndRegion(receiver, locatorPort);
-
- receiver
- .invoke("set up DistributionMessageObserver to 'hang' sender's put (on receiver)",
- () -> {
- final DistributionMessageObserver observer =
- new DistributionMessageObserver() {
-
- @Override
- public void beforeProcessMessage(final ClusterDistributionManager dm,
- final DistributionMessage message) {
- guardMessageProcessingHook(message, () -> {
- try {
- blackboard.signalGate(UPDATE_ENTERED_GATE);
- blackboard.waitForGate(SUSPEND_UPDATE_GATE);
- } catch (TimeoutException | InterruptedException e) {
- fail("message observus interruptus");
- }
- logger.info("BGB: got before process message: " + message);
- });
- }
- };
- DistributionMessageObserver.setInstance(observer);
- });
-
- final AsyncInvocation<Object> putInvocation = sender.invokeAsync("try a put", () -> {
- final Region<Object, Object> region = cache.getRegion(regionName);
- // test is going to close the cache while we are waiting for our ack
- assertThatThrownBy(() -> {
- region.put("hello", "world");
- }).isInstanceOf(DistributedSystemDisconnectedException.class);
- });
-
- // wait until our message observer is blocked
- blackboard.waitForGate(UPDATE_ENTERED_GATE);
-
- // at this point our put() is blocked waiting for a direct ack
- assertThat(putInvocation.isAlive()).as("put is waiting for remote region to ack").isTrue();
-
- /*
- * Now close the cache. The point of calling it is to test that we don't block while trying
- * to close connections. Cache.close() calls DistributedSystem.disconnect() which in turn
- * closes all the connections (and their sockets.) We want the sockets to close because that'll
- * cause our hung put() to see a DistributedSystemDisconnectedException.
- */
- sender.invoke("", () -> cache.close());
-
- // wait for put task to complete: with an exception, that is!
- putInvocation.get();
-
- // un-stick our message observer
- blackboard.signalGate(SUSPEND_UPDATE_GATE);
- }
-
- private void guardMessageProcessingHook(final DistributionMessage message,
- final Runnable runnable) {
- if (message instanceof UpdateMessage) {
- final UpdateMessage updateMessage = (UpdateMessage) message;
- if (updateMessage.getRegionPath().equals("/" + regionName)) {
- runnable.run();
- }
- }
- }
-
- private int createLocator(VM memberVM) {
- return memberVM.invoke("create locator", () -> {
- // if you need to debug SSL communications use this property:
- // System.setProperty("javax.net.debug", "all");
- System.setProperty(GMSJoinLeave.BYPASS_DISCOVERY_PROPERTY, "true");
- return Locator.startLocatorAndDS(0, new File(""), getDistributedSystemProperties())
- .getPort();
- });
- }
-
- private void createCacheAndRegion(VM memberVM, int locatorPort) {
- memberVM.invoke("start cache and create region", () -> {
- cache = createCache(locatorPort);
- cache.createRegionFactory(RegionShortcut.REPLICATE).create(regionName);
- });
- }
-
- private Cache createCache(int locatorPort) {
- // if you need to debug SSL communications use this property:
- // System.setProperty("javax.net.debug", "all");
- Properties properties = getDistributedSystemProperties();
- properties.setProperty(LOCATORS, "localhost[" + locatorPort + "]");
- return new CacheFactory(properties).create();
- }
-
- private Properties getDistributedSystemProperties() {
- Properties properties = new Properties();
- properties.setProperty(ENABLE_CLUSTER_CONFIGURATION, "false");
- properties.setProperty(USE_CLUSTER_CONFIGURATION, "false");
- properties.setProperty(NAME, "vm" + VM.getCurrentVMNum());
- properties.setProperty(CONSERVE_SOCKETS, "false"); // we are testing direct ack
- properties.setProperty(SOCKET_LEASE_TIME, "10000");
- properties.setProperty(SOCKET_BUFFER_SIZE, "" + SMALL_BUFFER_SIZE);
-
- properties.setProperty(SSL_ENABLED_COMPONENTS, "cluster,locator");
- properties
- .setProperty(SSL_KEYSTORE, createTempFileFromResource(getClass(), "server.keystore")
- .getAbsolutePath());
- properties.setProperty(SSL_TRUSTSTORE,
- createTempFileFromResource(getClass(), "server.keystore")
- .getAbsolutePath());
- properties.setProperty(SSL_PROTOCOLS, "TLSv1.2");
- properties.setProperty(SSL_KEYSTORE_PASSWORD, "password");
- properties.setProperty(SSL_TRUSTSTORE_PASSWORD, "password");
- properties.setProperty(SSL_REQUIRE_AUTHENTICATION, "true");
- return properties;
- }
-
-}
diff --git a/geode-core/src/distributedTest/resources/org/apache/geode/internal/tcp/server.keystore b/geode-core/src/distributedTest/resources/org/apache/geode/internal/tcp/server.keystore
deleted file mode 100644
index 8b5305f..0000000
Binary files a/geode-core/src/distributedTest/resources/org/apache/geode/internal/tcp/server.keystore and /dev/null differ
diff --git a/geode-core/src/integrationTest/java/org/apache/geode/internal/net/SSLSocketHostNameVerificationIntegrationTest.java b/geode-core/src/integrationTest/java/org/apache/geode/internal/net/SSLSocketHostNameVerificationIntegrationTest.java
index a70f3b1..dc7df44 100755
--- a/geode-core/src/integrationTest/java/org/apache/geode/internal/net/SSLSocketHostNameVerificationIntegrationTest.java
+++ b/geode-core/src/integrationTest/java/org/apache/geode/internal/net/SSLSocketHostNameVerificationIntegrationTest.java
@@ -215,9 +215,7 @@ public class SSLSocketHostNameVerificationIntegrationTest {
final NioSslEngine nioSslEngine = engine;
engine.close(socket.getChannel());
assertThatThrownBy(() -> {
- try (final ByteBufferSharing unused =
- nioSslEngine.unwrap(ByteBuffer.wrap(new byte[0]))) {
- }
+ nioSslEngine.unwrap(ByteBuffer.wrap(new byte[0]));
})
.isInstanceOf(IOException.class);
}
diff --git a/geode-core/src/integrationTest/java/org/apache/geode/internal/net/SSLSocketIntegrationTest.java b/geode-core/src/integrationTest/java/org/apache/geode/internal/net/SSLSocketIntegrationTest.java
index add6b9a..19eab4f 100755
--- a/geode-core/src/integrationTest/java/org/apache/geode/internal/net/SSLSocketIntegrationTest.java
+++ b/geode-core/src/integrationTest/java/org/apache/geode/internal/net/SSLSocketIntegrationTest.java
@@ -256,13 +256,11 @@ public class SSLSocketIntegrationTest {
ByteBuffer buffer = bbos.getContentBuffer();
System.out.println(
"client buffer position is " + buffer.position() + " and limit is " + buffer.limit());
- try (final ByteBufferSharing outputSharing = engine.wrap(buffer)) {
- ByteBuffer wrappedBuffer = outputSharing.getBuffer();
- System.out.println("client wrapped buffer position is " + wrappedBuffer.position()
- + " and limit is " + wrappedBuffer.limit());
- int bytesWritten = clientChannel.write(wrappedBuffer);
- System.out.println("client bytes written is " + bytesWritten);
- }
+ ByteBuffer wrappedBuffer = engine.wrap(buffer);
+ System.out.println("client wrapped buffer position is " + wrappedBuffer.position()
+ + " and limit is " + wrappedBuffer.limit());
+ int bytesWritten = clientChannel.write(wrappedBuffer);
+ System.out.println("client bytes written is " + bytesWritten);
}
private Thread startServerNIO(final ServerSocket serverSocket, int timeoutMillis)
@@ -301,9 +299,7 @@ public class SSLSocketIntegrationTest {
final NioSslEngine nioSslEngine = engine;
engine.close(socket.getChannel());
assertThatThrownBy(() -> {
- try (final ByteBufferSharing unused =
- nioSslEngine.unwrap(ByteBuffer.wrap(new byte[0]))) {
- }
+ nioSslEngine.unwrap(ByteBuffer.wrap(new byte[0]));
})
.isInstanceOf(IOException.class);
}
@@ -317,35 +313,24 @@ public class SSLSocketIntegrationTest {
private void readMessageFromNIOSSLClient(Socket socket, ByteBuffer buffer, NioSslEngine engine)
throws IOException {
- try (final ByteBufferSharing sharedBuffer = engine.getUnwrappedBuffer()) {
- final ByteBuffer unwrapped = sharedBuffer.getBuffer();
- // if we already have unencrypted data skip unwrapping
- if (unwrapped.position() == 0) {
- int bytesRead;
- // if we already have encrypted data skip reading from the socket
- if (buffer.position() == 0) {
- bytesRead = socket.getChannel().read(buffer);
- buffer.flip();
- } else {
- bytesRead = buffer.remaining();
- }
- System.out.println("server bytes read is " + bytesRead + ": buffer position is "
- + buffer.position() + " and limit is " + buffer.limit());
- try (final ByteBufferSharing sharedBuffer2 = engine.unwrap(buffer)) {
- final ByteBuffer unwrapped2 = sharedBuffer2.getBuffer();
-
- unwrapped2.flip();
- System.out.println("server unwrapped buffer position is " + unwrapped2.position()
- + " and limit is " + unwrapped2.limit());
- finishReadMessageFromNIOSSLClient(unwrapped2);
- }
+ ByteBuffer unwrapped = engine.getUnwrappedBuffer(buffer);
+ // if we already have unencrypted data skip unwrapping
+ if (unwrapped.position() == 0) {
+ int bytesRead;
+ // if we already have encrypted data skip reading from the socket
+ if (buffer.position() == 0) {
+ bytesRead = socket.getChannel().read(buffer);
+ buffer.flip();
} else {
- finishReadMessageFromNIOSSLClient(unwrapped);
+ bytesRead = buffer.remaining();
}
+ System.out.println("server bytes read is " + bytesRead + ": buffer position is "
+ + buffer.position() + " and limit is " + buffer.limit());
+ unwrapped = engine.unwrap(buffer);
+ unwrapped.flip();
+ System.out.println("server unwrapped buffer position is " + unwrapped.position()
+ + " and limit is " + unwrapped.limit());
}
- }
-
- private void finishReadMessageFromNIOSSLClient(final ByteBuffer unwrapped) throws IOException {
ByteBufferInputStream bbis = new ByteBufferInputStream(unwrapped);
DataInputStream dis = new DataInputStream(bbis);
String welcome = dis.readUTF();
diff --git a/geode-core/src/integrationTest/resources/org/apache/geode/codeAnalysis/excludedClasses.txt b/geode-core/src/integrationTest/resources/org/apache/geode/codeAnalysis/excludedClasses.txt
index 33f43c3..a46d5fc 100644
--- a/geode-core/src/integrationTest/resources/org/apache/geode/codeAnalysis/excludedClasses.txt
+++ b/geode-core/src/integrationTest/resources/org/apache/geode/codeAnalysis/excludedClasses.txt
@@ -104,4 +104,3 @@ org/apache/geode/cache/query/internal/xml/ElementType
org/apache/geode/cache/query/internal/xml/ElementType$1
org/apache/geode/cache/query/internal/xml/ElementType$2
org/apache/geode/cache/query/internal/xml/ElementType$3
-org/apache/geode/internal/net/ByteBufferSharingImpl$OpenAttemptTimedOut
\ No newline at end of file
diff --git a/geode-core/src/main/java/org/apache/geode/internal/net/ByteBufferSharing.java b/geode-core/src/main/java/org/apache/geode/internal/net/ByteBufferSharing.java
deleted file mode 100644
index cdfa897..0000000
--- a/geode-core/src/main/java/org/apache/geode/internal/net/ByteBufferSharing.java
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * 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.internal.net;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-
-
-/**
- * When a {@link ByteBufferSharing} is acquired in a try-with-resources the buffer is available (for
- * reading and modification) within the scope of that try block.
- *
- * Releases managed ByteBuffer back to pool after last reference is dropped.
- */
-public interface ByteBufferSharing extends AutoCloseable {
-
- /**
- * Call this method only within a try-with-resource in which this {@link ByteBufferSharing} was
- * acquired. Retain the reference only within the scope of that try-with-resources.
- *
- * @return the buffer: manipulable only within the scope of the try-with-resources
- * @throws IOException if the buffer is no longer accessible
- */
- ByteBuffer getBuffer() throws IOException;
-
- /**
- * Expand the buffer if needed. This may return a different object so be sure to pay attention to
- * the return value if you need access to the potentially- expanded buffer.
- *
- * Subsequent calls to {@link #getBuffer()} will return that new buffer too.
- *
- * @return the same buffer or a different (bigger) buffer
- * @throws IOException if the buffer is no longer accessible
- */
- ByteBuffer expandWriteBufferIfNeeded(final int newCapacity) throws IOException;
-
- /**
- * Override {@link AutoCloseable#close()} without throws clause since we don't need one.
- */
- @Override
- void close();
-}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/net/ByteBufferSharingImpl.java b/geode-core/src/main/java/org/apache/geode/internal/net/ByteBufferSharingImpl.java
deleted file mode 100644
index e9a941e..0000000
--- a/geode-core/src/main/java/org/apache/geode/internal/net/ByteBufferSharingImpl.java
+++ /dev/null
@@ -1,148 +0,0 @@
-/*
- * 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.internal.net;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-
-import org.apache.geode.annotations.VisibleForTesting;
-import org.apache.geode.internal.net.BufferPool.BufferType;
-
-/**
- * An {@link AutoCloseable} meant to be acquired in a try-with-resources statement. The resource (a
- * {@link ByteBuffer}) is available (for reading and modification) in the scope of the
- * try-with-resources.
- */
-class ByteBufferSharingImpl implements ByteBufferSharing {
-
- static class OpenAttemptTimedOut extends Exception {
- }
-
- private final Lock lock;
- private final AtomicBoolean isClosed;
- // mutable because in general our ByteBuffer may need to be resized (grown or compacted)
- private ByteBuffer buffer;
- private final BufferType bufferType;
- private final AtomicInteger counter;
- private final BufferPool bufferPool;
-
- /**
- * This constructor is for use only by the owner of the shared resource (a {@link ByteBuffer}).
- *
- * A resource owner must invoke {@link #open()} once for each reference that escapes (is passed
- * to an external object or is returned to an external caller.)
- *
- * This constructor acquires no lock. The reference count will be 1 after this constructor
- * completes.
- */
- ByteBufferSharingImpl(final ByteBuffer buffer, final BufferType bufferType,
- final BufferPool bufferPool) {
- this.buffer = buffer;
- this.bufferType = bufferType;
- this.bufferPool = bufferPool;
- lock = new ReentrantLock();
- counter = new AtomicInteger(1);
- isClosed = new AtomicBoolean(false);
- }
-
- /**
- * The destructor. Called by the resource owner to undo the work of the constructor.
- */
- void destruct() {
- if (isClosed.compareAndSet(false, true)) {
- dropReference();
- }
- }
-
- /**
- * This method is for use only by the owner of the shared resource. It's used for handing out
- * references to the shared resource. So it does reference counting and also acquires a lock.
- *
- * Resource owners call this method as the last thing before returning a reference to the caller.
- * That caller binds that reference to a variable in a try-with-resources statement and relies on
- * the AutoCloseable protocol to invoke {@link #close()} on the object at the end of the block.
- */
- ByteBufferSharing open() {
- lock.lock();
- addReference();
- return this;
- }
-
- /**
- * This variant throws {@link OpenAttemptTimedOut} if it can't acquire the lock in time.
- */
- ByteBufferSharing open(final long time, final TimeUnit unit) throws OpenAttemptTimedOut {
- try {
- if (!lock.tryLock(time, unit)) {
- throw new OpenAttemptTimedOut();
- }
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new OpenAttemptTimedOut();
- }
- addReference();
- return this;
- }
-
- @Override
- public ByteBuffer getBuffer() throws IOException {
- if (isClosed.get()) {
- throw new IOException("NioSslEngine has been closed");
- } else {
- return buffer;
- }
- }
-
- @Override
- public ByteBuffer expandWriteBufferIfNeeded(final int newCapacity) throws IOException {
- return buffer = bufferPool.expandWriteBufferIfNeeded(bufferType, getBuffer(), newCapacity);
- }
-
- @Override
- public void close() {
- /*
- * We are counting on our ReentrantLock throwing an exception if the current thread
- * does not hold the lock. In that case dropReference() will not be called. This
- * prevents ill-behaved clients (clients that call close() too many times) from
- * corrupting our reference count.
- */
- lock.unlock();
- dropReference();
- }
-
- private int addReference() {
- return counter.incrementAndGet();
- }
-
- private int dropReference() {
- final int usages = counter.decrementAndGet();
- if (usages == 0) {
- bufferPool.releaseBuffer(bufferType, buffer);
- }
- return usages;
- }
-
- @VisibleForTesting
- public void setBufferForTestingOnly(final ByteBuffer newBufferForTesting) {
- buffer = newBufferForTesting;
- }
-
-}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/net/ByteBufferSharingNoOp.java b/geode-core/src/main/java/org/apache/geode/internal/net/ByteBufferSharingNoOp.java
deleted file mode 100644
index bd707e3..0000000
--- a/geode-core/src/main/java/org/apache/geode/internal/net/ByteBufferSharingNoOp.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * 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.internal.net;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-
-/**
- * An {@link AutoCloseable} meant to be acquired in a try-with-resources statement. The resource (a
- * {@link ByteBuffer}) is available (for reading and modification) in the scope of the
- * try-with-resources.
- *
- * This implementation is a "no-op". It performs no actual locking and no reference counting. It's
- * meant for use with the {@link NioPlainEngine} only, since that engine keeps no buffers and so,
- * needs no reference counting on buffers, nor any synchronization around access to buffers.
- *
- * See also {@link ByteBufferSharingImpl}
- */
-class ByteBufferSharingNoOp implements ByteBufferSharing {
-
- private final ByteBuffer buffer;
-
- ByteBufferSharingNoOp(final ByteBuffer buffer) {
- this.buffer = buffer;
- }
-
- @Override
- public ByteBuffer getBuffer() {
- return buffer;
- }
-
- @Override
- public ByteBuffer expandWriteBufferIfNeeded(final int newCapacity) throws IOException {
- throw new UnsupportedOperationException("Can't expand buffer when using NioPlainEngine");
- }
-
- @Override
- public void close() {}
-}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/net/NioFilter.java b/geode-core/src/main/java/org/apache/geode/internal/net/NioFilter.java
index eb53f0e..9c437ad 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/net/NioFilter.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/net/NioFilter.java
@@ -19,53 +19,47 @@ import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
/**
- * Prior to transmitting a buffer or processing a received buffer a NioFilter should be called to
- * wrap (transmit) or unwrap (received) the buffer in case SSL is being used.<br>
- * Implementations of
- * this class may not be thread-safe in regard to the buffers their methods return. These may be
- * internal state that, if used concurrently by multiple threads could cause corruption. Appropriate
- * external synchronization must be used in order to provide thread-safety. Do this by invoking
- * getSynchObject() and synchronizing on the returned object while using the buffer.
+ * Prior to transmitting a buffer or processing a received buffer
+ * a NioFilter should be called to wrap (transmit) or unwrap (received)
+ * the buffer in case SSL is being used.<br>
+ * Implementations of this class may not be thread-safe in regard to
+ * the buffers their methods return. These may be internal state that,
+ * if used concurrently by multiple threads could cause corruption.
+ * Appropriate external synchronization must be used in order to provide
+ * thread-safety. Do this by invoking getSynchObject() and synchronizing on
+ * the returned object while using the buffer.
*/
public interface NioFilter {
/**
* wrap bytes for transmission to another process
- *
- * Be sure to call close() on the returned {@link ByteBufferSharing}. The best way to do that is
- * to call this method in a try-with-resources statement.
*/
- ByteBufferSharing wrap(ByteBuffer buffer) throws IOException;
+ ByteBuffer wrap(ByteBuffer buffer) throws IOException;
/**
- * unwrap bytes received from another process. The unwrapped buffer should be flipped before
- * reading. When done reading invoke doneReading() to reset for future read ops
- *
- * Be sure to call close() on the returned {@link ByteBufferSharing}. The best way to do that is
- * to call this method in a try-with-resources statement.
+ * unwrap bytes received from another process. The unwrapped
+ * buffer should be flipped before reading. When done reading invoke
+ * doneReading() to reset for future read ops
*/
- ByteBufferSharing unwrap(ByteBuffer wrappedBuffer) throws IOException;
+ ByteBuffer unwrap(ByteBuffer wrappedBuffer) throws IOException;
/**
- * ensure that the wrapped buffer has enough room to read the given amount of data. This must be
- * invoked before readAtLeast. A new buffer may be returned by this method.
+ * ensure that the wrapped buffer has enough room to read the given amount of data.
+ * This must be invoked before readAtLeast. A new buffer may be returned by this method.
*/
ByteBuffer ensureWrappedCapacity(int amount, ByteBuffer wrappedBuffer,
BufferPool.BufferType bufferType);
/**
- * read at least the indicated amount of bytes from the given socket. The buffer position will be
- * ready for reading the data when this method returns. Note: you must invoke
- * ensureWrappedCapacity with the given amount prior to each invocation of this method.
+ * read at least the indicated amount of bytes from the given
+ * socket. The buffer position will be ready for reading
+ * the data when this method returns. Note: you must invoke ensureWrappedCapacity
+ * with the given amount prior to each invocation of this method.
* <br>
* wrappedBuffer = filter.ensureWrappedCapacity(amount, wrappedBuffer, etc.);<br>
- * unwrappedBuffer
- * = filter.readAtLeast(channel, amount, wrappedBuffer, etc.)
- *
- * Be sure to call close() on the returned {@link ByteBufferSharing}. The best way to do that is
- * to call this method in a try-with-resources statement.
+ * unwrappedBuffer = filter.readAtLeast(channel, amount, wrappedBuffer, etc.)
*/
- ByteBufferSharing readAtLeast(SocketChannel channel, int amount, ByteBuffer wrappedBuffer)
+ ByteBuffer readAtLeast(SocketChannel channel, int amount, ByteBuffer wrappedBuffer)
throws IOException;
/**
@@ -87,19 +81,28 @@ public interface NioFilter {
}
}
+ default boolean isClosed() {
+ return false;
+ }
+
/**
* invoke this method when you are done using the NioFilter
+ *
*/
default void close(SocketChannel socketChannel) {
// nothing by default
}
/**
- * Returns the sharing object for the {@link NioFilter}'s unwrapped buffer, if one exists.
- *
- * Be sure to call close() on the returned {@link ByteBufferSharing}. The best way to do that is
- * to call this method in a try-with-resources statement.
+ * returns the unwrapped byte buffer associated with the given wrapped buffer.
*/
- ByteBufferSharing getUnwrappedBuffer();
+ ByteBuffer getUnwrappedBuffer(ByteBuffer wrappedBuffer);
+ /**
+ * returns an object to be used in synchronizing on the use of buffers returned by
+ * a NioFilter.
+ */
+ default Object getSynchObject() {
+ return this;
+ }
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/net/NioPlainEngine.java b/geode-core/src/main/java/org/apache/geode/internal/net/NioPlainEngine.java
index 8b5df96..3ebce38 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/net/NioPlainEngine.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/net/NioPlainEngine.java
@@ -20,7 +20,6 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
-import org.apache.geode.annotations.internal.MakeImmutable;
import org.apache.geode.internal.Assert;
/**
@@ -28,12 +27,6 @@ import org.apache.geode.internal.Assert;
* secure communications.
*/
public class NioPlainEngine implements NioFilter {
-
- // this variable requires the MakeImmutable annotation but the buffer is empty and
- // not really modifiable
- @MakeImmutable
- private static final ByteBuffer EMPTY_BUFFER = ByteBuffer.allocate(0);
-
private final BufferPool bufferPool;
int lastReadPosition;
@@ -45,14 +38,14 @@ public class NioPlainEngine implements NioFilter {
}
@Override
- public ByteBufferSharing wrap(ByteBuffer buffer) {
- return shareBuffer(buffer);
+ public ByteBuffer wrap(ByteBuffer buffer) {
+ return buffer;
}
@Override
- public ByteBufferSharing unwrap(ByteBuffer wrappedBuffer) {
+ public ByteBuffer unwrap(ByteBuffer wrappedBuffer) {
wrappedBuffer.position(wrappedBuffer.limit());
- return shareBuffer(wrappedBuffer);
+ return wrappedBuffer;
}
@Override
@@ -89,7 +82,7 @@ public class NioPlainEngine implements NioFilter {
}
@Override
- public ByteBufferSharing readAtLeast(SocketChannel channel, int bytes, ByteBuffer wrappedBuffer)
+ public ByteBuffer readAtLeast(SocketChannel channel, int bytes, ByteBuffer wrappedBuffer)
throws IOException {
ByteBuffer buffer = wrappedBuffer;
@@ -115,7 +108,7 @@ public class NioPlainEngine implements NioFilter {
buffer.position(lastProcessedPosition);
lastProcessedPosition += bytes;
- return shareBuffer(buffer);
+ return buffer;
}
public void doneReading(ByteBuffer unwrappedBuffer) {
@@ -128,12 +121,8 @@ public class NioPlainEngine implements NioFilter {
}
@Override
- public ByteBufferSharing getUnwrappedBuffer() {
- return shareBuffer(EMPTY_BUFFER);
- }
-
- private ByteBufferSharingNoOp shareBuffer(final ByteBuffer wrappedBuffer) {
- return new ByteBufferSharingNoOp(wrappedBuffer);
+ public ByteBuffer getUnwrappedBuffer(ByteBuffer wrappedBuffer) {
+ return wrappedBuffer;
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/net/NioSslEngine.java b/geode-core/src/main/java/org/apache/geode/internal/net/NioSslEngine.java
index 7e642ce..6f32501 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/net/NioSslEngine.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/net/NioSslEngine.java
@@ -40,19 +40,24 @@ import javax.net.ssl.SSLSession;
import org.apache.logging.log4j.Logger;
import org.apache.geode.GemFireIOException;
-import org.apache.geode.annotations.VisibleForTesting;
+import org.apache.geode.annotations.internal.MakeImmutable;
import org.apache.geode.internal.net.BufferPool.BufferType;
-import org.apache.geode.internal.net.ByteBufferSharingImpl.OpenAttemptTimedOut;
import org.apache.geode.logging.internal.log4j.api.LogService;
/**
- * NioSslEngine uses an SSLEngine to bind SSL logic to a data source. This class is not thread safe.
- * Its use should be confined to one thread or should be protected by external synchronization.
+ * NioSslEngine uses an SSLEngine to bind SSL logic to a data source. This class is not thread
+ * safe. Its use should be confined to one thread or should be protected by external
+ * synchronization.
*/
public class NioSslEngine implements NioFilter {
private static final Logger logger = LogService.getLogger();
+ // this variable requires the MakeImmutable annotation but the buffer is empty and
+ // not really modifiable
+ @MakeImmutable
+ private static final ByteBuffer EMPTY_BUFFER = ByteBuffer.allocate(0);
+
private final BufferPool bufferPool;
private boolean closed;
@@ -60,28 +65,23 @@ public class NioSslEngine implements NioFilter {
SSLEngine engine;
/**
- * holds bytes wrapped by the SSLEngine; a.k.a. myNetData
+ * myNetData holds bytes wrapped by the SSLEngine
*/
- private final ByteBufferSharingImpl outputSharing;
+ ByteBuffer myNetData;
/**
- * holds the last unwrapped data from a peer; a.k.a. peerAppData
+ * peerAppData holds the last unwrapped data from a peer
*/
- private final ByteBufferSharingImpl inputSharing;
+ ByteBuffer peerAppData;
NioSslEngine(SSLEngine engine, BufferPool bufferPool) {
SSLSession session = engine.getSession();
int appBufferSize = session.getApplicationBufferSize();
int packetBufferSize = engine.getSession().getPacketBufferSize();
- closed = false;
this.engine = engine;
this.bufferPool = bufferPool;
- outputSharing =
- new ByteBufferSharingImpl(bufferPool.acquireDirectSenderBuffer(packetBufferSize),
- TRACKED_SENDER, bufferPool);
- inputSharing =
- new ByteBufferSharingImpl(bufferPool.acquireNonDirectReceiveBuffer(appBufferSize),
- TRACKED_RECEIVER, bufferPool);
+ this.myNetData = bufferPool.acquireDirectSenderBuffer(packetBufferSize);
+ this.peerAppData = bufferPool.acquireNonDirectReceiveBuffer(appBufferSize);
}
/**
@@ -135,65 +135,57 @@ public class NioSslEngine implements NioFilter {
switch (status) {
case NEED_UNWRAP:
- try (final ByteBufferSharing inputSharing = shareInputBuffer()) {
- final ByteBuffer peerAppData = inputSharing.getBuffer();
-
- // Receive handshaking data from peer
- int dataRead = socketChannel.read(handshakeBuffer);
-
- // Process incoming handshaking data
- handshakeBuffer.flip();
-
-
- engineResult = engine.unwrap(handshakeBuffer, peerAppData);
- handshakeBuffer.compact();
- status = engineResult.getHandshakeStatus();
-
- // if we're not finished, there's nothing to process and no data was read let's hang out
- // for a little
- if (peerAppData.remaining() == 0 && dataRead == 0 && status == NEED_UNWRAP) {
- Thread.sleep(10);
- }
+ // Receive handshaking data from peer
+ int dataRead = socketChannel.read(handshakeBuffer);
+
+ // Process incoming handshaking data
+ handshakeBuffer.flip();
+ engineResult = engine.unwrap(handshakeBuffer, peerAppData);
+ handshakeBuffer.compact();
+ status = engineResult.getHandshakeStatus();
+
+ // if we're not finished, there's nothing to process and no data was read let's hang out
+ // for a little
+ if (peerAppData.remaining() == 0 && dataRead == 0 && status == NEED_UNWRAP) {
+ Thread.sleep(10);
+ }
- if (engineResult.getStatus() == BUFFER_OVERFLOW) {
- inputSharing.expandWriteBufferIfNeeded(peerAppData.capacity() * 2);
- }
- break;
+ if (engineResult.getStatus() == BUFFER_OVERFLOW) {
+ peerAppData =
+ expandWriteBuffer(TRACKED_RECEIVER, peerAppData, peerAppData.capacity() * 2);
}
+ break;
case NEED_WRAP:
- try (final ByteBufferSharing outputSharing = shareOutputBuffer()) {
- final ByteBuffer myNetData = outputSharing.getBuffer();
-
- // Empty the local network packet buffer.
- myNetData.clear();
-
- // Generate handshaking data
- engineResult = engine.wrap(myAppData, myNetData);
- status = engineResult.getHandshakeStatus();
-
- // Check status
- switch (engineResult.getStatus()) {
- case BUFFER_OVERFLOW:
- // no need to assign return value because we will never reference it
- outputSharing.expandWriteBufferIfNeeded(myNetData.capacity() * 2);
- break;
- case OK:
- myNetData.flip();
- // Send the handshaking data to peer
- while (myNetData.hasRemaining()) {
- socketChannel.write(myNetData);
- }
- break;
- case CLOSED:
- break;
- default:
- logger.info("handshake terminated with illegal state due to {}", status);
- throw new IllegalStateException(
- "Unknown SSLEngineResult status: " + engineResult.getStatus());
- }
- break;
+ // Empty the local network packet buffer.
+ myNetData.clear();
+
+ // Generate handshaking data
+ engineResult = engine.wrap(myAppData, myNetData);
+ status = engineResult.getHandshakeStatus();
+
+ // Check status
+ switch (engineResult.getStatus()) {
+ case BUFFER_OVERFLOW:
+ myNetData =
+ expandWriteBuffer(TRACKED_SENDER, myNetData,
+ myNetData.capacity() * 2);
+ break;
+ case OK:
+ myNetData.flip();
+ // Send the handshaking data to peer
+ while (myNetData.hasRemaining()) {
+ socketChannel.write(myNetData);
+ }
+ break;
+ case CLOSED:
+ break;
+ default:
+ logger.info("handshake terminated with illegal state due to {}", status);
+ throw new IllegalStateException(
+ "Unknown SSLEngineResult status: " + engineResult.getStatus());
}
+ break;
case NEED_TASK:
// Handle blocking tasks
handleBlockingTasks();
@@ -221,6 +213,17 @@ public class NioSslEngine implements NioFilter {
return true;
}
+ ByteBuffer expandWriteBuffer(BufferType type, ByteBuffer existing,
+ int desiredCapacity) {
+ return bufferPool.expandWriteBufferIfNeeded(type, existing, desiredCapacity);
+ }
+
+ synchronized void checkClosed() throws IOException {
+ if (closed) {
+ throw new IOException("NioSslEngine has been closed");
+ }
+ }
+
void handleBlockingTasks() {
Runnable task;
while ((task = engine.getDelegatedTask()) != null) {
@@ -230,77 +233,72 @@ public class NioSslEngine implements NioFilter {
}
@Override
- public ByteBufferSharing wrap(ByteBuffer appData) throws IOException {
- try (final ByteBufferSharing outputSharing = shareOutputBuffer()) {
+ public synchronized ByteBuffer wrap(ByteBuffer appData) throws IOException {
+ checkClosed();
- ByteBuffer myNetData = outputSharing.getBuffer();
+ myNetData.clear();
- myNetData.clear();
+ while (appData.hasRemaining()) {
+ // ensure we have lots of capacity since encrypted data might
+ // be larger than the app data
+ int remaining = myNetData.capacity() - myNetData.position();
- while (appData.hasRemaining()) {
- // ensure we have lots of capacity since encrypted data might
- // be larger than the app data
- int remaining = myNetData.capacity() - myNetData.position();
-
- if (remaining < (appData.remaining() * 2)) {
- int newCapacity = expandedCapacity(appData, myNetData);
- myNetData = outputSharing.expandWriteBufferIfNeeded(newCapacity);
- }
+ if (remaining < (appData.remaining() * 2)) {
+ int newCapacity = expandedCapacity(appData, myNetData);
+ myNetData = expandWriteBuffer(TRACKED_SENDER, myNetData, newCapacity);
+ }
- SSLEngineResult wrapResult = engine.wrap(appData, myNetData);
+ SSLEngineResult wrapResult = engine.wrap(appData, myNetData);
- if (wrapResult.getHandshakeStatus() == NEED_TASK) {
- handleBlockingTasks();
- }
+ if (wrapResult.getHandshakeStatus() == NEED_TASK) {
+ handleBlockingTasks();
+ }
- if (wrapResult.getStatus() != OK) {
- throw new SSLException("Error encrypting data: " + wrapResult);
- }
+ if (wrapResult.getStatus() != OK) {
+ throw new SSLException("Error encrypting data: " + wrapResult);
}
+ }
- myNetData.flip();
+ myNetData.flip();
- return shareOutputBuffer();
- }
+ return myNetData;
}
@Override
- public ByteBufferSharing unwrap(ByteBuffer wrappedBuffer) throws IOException {
- try (final ByteBufferSharing inputSharing = shareInputBuffer()) {
-
- ByteBuffer peerAppData = inputSharing.getBuffer();
-
- // note that we do not clear peerAppData as it may hold a partial
- // message. TcpConduit, for instance, uses message chunking to
- // transmit large payloads and we may have read a partial chunk
- // during the previous unwrap
-
- peerAppData.limit(peerAppData.capacity());
- while (wrappedBuffer.hasRemaining()) {
- SSLEngineResult unwrapResult = engine.unwrap(wrappedBuffer, peerAppData);
- switch (unwrapResult.getStatus()) {
- case BUFFER_OVERFLOW:
- // buffer overflow expand and try again - double the available decryption space
- int newCapacity =
- (peerAppData.capacity() - peerAppData.position()) * 2 + peerAppData.position();
- newCapacity = Math.max(newCapacity, peerAppData.capacity() / 2 * 3);
- peerAppData = inputSharing.expandWriteBufferIfNeeded(newCapacity);
- peerAppData.limit(peerAppData.capacity());
- break;
- case BUFFER_UNDERFLOW:
- // partial data - need to read more. When this happens the SSLEngine will not have
- // changed the buffer position
- wrappedBuffer.compact();
- return shareInputBuffer();
- case OK:
- break;
- default:
- throw new SSLException("Error decrypting data: " + unwrapResult);
- }
+ public synchronized ByteBuffer unwrap(ByteBuffer wrappedBuffer) throws IOException {
+ checkClosed();
+
+ // note that we do not clear peerAppData as it may hold a partial
+ // message. TcpConduit, for instance, uses message chunking to
+ // transmit large payloads and we may have read a partial chunk
+ // during the previous unwrap
+
+ peerAppData.limit(peerAppData.capacity());
+ while (wrappedBuffer.hasRemaining()) {
+ SSLEngineResult unwrapResult = engine.unwrap(wrappedBuffer, peerAppData);
+ switch (unwrapResult.getStatus()) {
+ case BUFFER_OVERFLOW:
+ // buffer overflow expand and try again - double the available decryption space
+ int newCapacity =
+ (peerAppData.capacity() - peerAppData.position()) * 2 + peerAppData.position();
+ newCapacity = Math.max(newCapacity, peerAppData.capacity() / 2 * 3);
+ peerAppData =
+ bufferPool.expandWriteBufferIfNeeded(TRACKED_RECEIVER, peerAppData, newCapacity);
+ peerAppData.limit(peerAppData.capacity());
+ break;
+ case BUFFER_UNDERFLOW:
+ // partial data - need to read more. When this happens the SSLEngine will not have
+ // changed the buffer position
+ wrappedBuffer.compact();
+ return peerAppData;
+ case OK:
+ break;
+ default:
+ throw new SSLException("Error decrypting data: " + unwrapResult);
}
- wrappedBuffer.clear();
- return shareInputBuffer();
}
+ wrappedBuffer.clear();
+ return peerAppData;
}
@Override
@@ -317,45 +315,50 @@ public class NioSslEngine implements NioFilter {
}
@Override
- public ByteBufferSharing readAtLeast(SocketChannel channel, int bytes,
+ public ByteBuffer readAtLeast(SocketChannel channel, int bytes,
ByteBuffer wrappedBuffer) throws IOException {
- try (final ByteBufferSharing inputSharing = shareInputBuffer()) {
-
- ByteBuffer peerAppData = inputSharing.getBuffer();
-
- if (peerAppData.capacity() > bytes) {
- // we already have a buffer that's big enough
- if (peerAppData.capacity() - peerAppData.position() < bytes) {
- peerAppData.compact();
- peerAppData.flip();
- }
+ if (peerAppData.capacity() > bytes) {
+ // we already have a buffer that's big enough
+ if (peerAppData.capacity() - peerAppData.position() < bytes) {
+ peerAppData.compact();
+ peerAppData.flip();
}
+ }
- while (peerAppData.remaining() < bytes) {
- wrappedBuffer.limit(wrappedBuffer.capacity());
- int amountRead = channel.read(wrappedBuffer);
- if (amountRead < 0) {
- throw new EOFException();
- }
- if (amountRead > 0) {
- wrappedBuffer.flip();
- // prep the decoded buffer for writing
- peerAppData.compact();
- try (final ByteBufferSharing inputSharing2 = unwrap(wrappedBuffer)) {
- // done writing to the decoded buffer - prep it for reading again
- final ByteBuffer peerAppDataNew = inputSharing2.getBuffer();
- peerAppDataNew.flip();
- peerAppData = peerAppDataNew; // loop needs new reference!
- }
- }
+ while (peerAppData.remaining() < bytes) {
+ wrappedBuffer.limit(wrappedBuffer.capacity());
+ int amountRead = channel.read(wrappedBuffer);
+ if (amountRead < 0) {
+ throw new EOFException();
+ }
+ if (amountRead > 0) {
+ wrappedBuffer.flip();
+ // prep the decoded buffer for writing
+ peerAppData.compact();
+ peerAppData = unwrap(wrappedBuffer);
+ // done writing to the decoded buffer - prep it for reading again
+ peerAppData.flip();
}
- return shareInputBuffer();
}
+ return peerAppData;
}
@Override
- public ByteBufferSharing getUnwrappedBuffer() {
- return shareInputBuffer();
+ public ByteBuffer getUnwrappedBuffer(ByteBuffer wrappedBuffer) {
+ return peerAppData;
+ }
+
+ /**
+ * ensures that the unwrapped buffer associated with the given wrapped buffer has
+ * sufficient capacity for the given amount of bytes. This may compact the
+ * buffer or it may return a new buffer.
+ */
+ public ByteBuffer ensureUnwrappedCapacity(int amount) {
+ // for TTLS the app-data buffers do not need to be tracked direct-buffers since we
+ // do not use them for I/O operations
+ peerAppData =
+ bufferPool.expandReadBufferIfNeeded(TRACKED_RECEIVER, peerAppData, amount);
+ return peerAppData;
}
@Override
@@ -366,14 +369,16 @@ public class NioSslEngine implements NioFilter {
}
@Override
+ public synchronized boolean isClosed() {
+ return closed;
+ }
+
+ @Override
public synchronized void close(SocketChannel socketChannel) {
if (closed) {
return;
}
- closed = true;
- inputSharing.destruct();
- try (final ByteBufferSharing outputSharing = shareOutputBuffer(1, TimeUnit.MINUTES)) {
- final ByteBuffer myNetData = outputSharing.getBuffer();
+ try {
if (!engine.isOutboundDone()) {
ByteBuffer empty = ByteBuffer.wrap(new byte[0]);
@@ -400,13 +405,14 @@ public class NioSslEngine implements NioFilter {
// we can't send a close message if the channel is closed
} catch (IOException e) {
throw new GemFireIOException("exception closing SSL session", e);
- } catch (final OpenAttemptTimedOut _unused) {
- logger.info(String.format("Couldn't get output lock in time, eliding TLS close message"));
- if (!engine.isOutboundDone()) {
- engine.closeOutbound();
- }
} finally {
- outputSharing.destruct();
+ ByteBuffer netData = myNetData;
+ ByteBuffer appData = peerAppData;
+ myNetData = null;
+ peerAppData = EMPTY_BUFFER;
+ bufferPool.releaseBuffer(TRACKED_SENDER, netData);
+ bufferPool.releaseBuffer(TRACKED_RECEIVER, appData);
+ this.closed = true;
}
}
@@ -415,17 +421,4 @@ public class NioSslEngine implements NioFilter {
targetBuffer.capacity() * 2);
}
- @VisibleForTesting
- public ByteBufferSharing shareOutputBuffer() {
- return outputSharing.open();
- }
-
- private ByteBufferSharing shareOutputBuffer(final long time, final TimeUnit unit)
- throws OpenAttemptTimedOut {
- return outputSharing.open(time, unit);
- }
-
- public ByteBufferSharing shareInputBuffer() {
- return inputSharing.open();
- }
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/tcp/Connection.java b/geode-core/src/main/java/org/apache/geode/internal/tcp/Connection.java
index c6619d1..9292727 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/tcp/Connection.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/tcp/Connection.java
@@ -78,7 +78,6 @@ import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.SystemTimer;
import org.apache.geode.internal.SystemTimer.SystemTimerTask;
import org.apache.geode.internal.net.BufferPool;
-import org.apache.geode.internal.net.ByteBufferSharing;
import org.apache.geode.internal.net.NioFilter;
import org.apache.geode.internal.net.NioPlainEngine;
import org.apache.geode.internal.net.SocketCreator;
@@ -801,12 +800,11 @@ public class Connection implements Runnable {
@VisibleForTesting
void clearSSLInputBuffer() {
if (getConduit().useSSL() && ioFilter != null) {
- try (final ByteBufferSharing sharedBuffer = ioFilter.getUnwrappedBuffer()) {
- // clear out any remaining handshake bytes
- try {
- sharedBuffer.getBuffer().position(0).limit(0);
- } catch (IOException e) {
- // means the NioFilter was already closed
+ synchronized (ioFilter.getSynchObject()) {
+ if (!ioFilter.isClosed()) {
+ // clear out any remaining handshake bytes
+ ByteBuffer buffer = ioFilter.getUnwrappedBuffer(inputBuffer);
+ buffer.position(0).limit(0);
}
}
}
@@ -2452,9 +2450,8 @@ public class Connection implements Runnable {
long queueTimeoutTarget = now + asyncQueueTimeout;
channel.configureBlocking(false);
try {
- try (final ByteBufferSharing outputSharing = ioFilter.wrap(buffer)) {
- final ByteBuffer wrappedBuffer = outputSharing.getBuffer();
-
+ synchronized (ioFilter.getSynchObject()) {
+ ByteBuffer wrappedBuffer = ioFilter.wrap(buffer);
int waitTime = 1;
do {
owner.getConduit().getCancelCriterion().checkCancelInProgress(null);
@@ -2607,9 +2604,9 @@ public class Connection implements Runnable {
}
// fall through
}
- try (final ByteBufferSharing outputSharing = ioFilter.wrap(buffer)) {
- final ByteBuffer wrappedBuffer = outputSharing.getBuffer();
-
+ // synchronize on the ioFilter while using its network buffer
+ synchronized (ioFilter.getSynchObject()) {
+ ByteBuffer wrappedBuffer = ioFilter.wrap(buffer);
while (wrappedBuffer.remaining() > 0) {
int amtWritten = 0;
long start = stats.startSocketWrite(true);
@@ -2661,12 +2658,10 @@ public class Connection implements Runnable {
final Version version = getRemoteVersion();
try {
msgReader = new MsgReader(this, ioFilter, version);
-
ReplyMessage msg;
int len;
- // (we have to lock here to protect between reading header and message body)
- try (final ByteBufferSharing _unused = ioFilter.getUnwrappedBuffer()) {
+ synchronized (ioFilter.getSynchObject()) {
Header header = msgReader.readHeader();
if (header.getMessageType() == NORMAL_MSG_TYPE) {
@@ -2683,7 +2678,7 @@ public class Connection implements Runnable {
releaseMsgDestreamer(header.getMessageId(), destreamer);
len = destreamer.size();
}
- }
+ } // sync
// I'd really just like to call dispatchMessage here. However,
// that call goes through a bunch of checks that knock about
// 10% of the performance. Since this direct-ack stuff is all
@@ -2750,9 +2745,8 @@ public class Connection implements Runnable {
private void processInputBuffer() throws ConnectionException, IOException {
inputBuffer.flip();
- try (final ByteBufferSharing sharedBuffer = ioFilter.unwrap(inputBuffer)) {
- final ByteBuffer peerDataBuffer = sharedBuffer.getBuffer();
-
+ synchronized (ioFilter.getSynchObject()) {
+ ByteBuffer peerDataBuffer = ioFilter.unwrap(inputBuffer);
peerDataBuffer.flip();
boolean done = false;
diff --git a/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgReader.java b/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgReader.java
index 503e48b..396ece2 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgReader.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgReader.java
@@ -26,7 +26,6 @@ import org.apache.geode.distributed.internal.ReplyProcessor21;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.net.BufferPool;
-import org.apache.geode.internal.net.ByteBufferSharing;
import org.apache.geode.internal.net.NioFilter;
import org.apache.geode.internal.serialization.Version;
import org.apache.geode.logging.internal.log4j.api.LogService;
@@ -55,8 +54,8 @@ public class MsgReader {
}
Header readHeader() throws IOException {
- try (final ByteBufferSharing sharedBuffer = readAtLeast(Connection.MSG_HEADER_BYTES)) {
- ByteBuffer unwrappedBuffer = sharedBuffer.getBuffer();
+ synchronized (ioFilter.getSynchObject()) {
+ ByteBuffer unwrappedBuffer = readAtLeast(Connection.MSG_HEADER_BYTES);
Assert.assertTrue(unwrappedBuffer.remaining() >= Connection.MSG_HEADER_BYTES);
@@ -90,8 +89,8 @@ public class MsgReader {
*/
DistributionMessage readMessage(Header header)
throws IOException, ClassNotFoundException {
- try (final ByteBufferSharing sharedBuffer = readAtLeast(header.messageLength)) {
- ByteBuffer nioInputBuffer = sharedBuffer.getBuffer();
+ synchronized (ioFilter.getSynchObject()) {
+ ByteBuffer nioInputBuffer = readAtLeast(header.messageLength);
Assert.assertTrue(nioInputBuffer.remaining() >= header.messageLength);
this.getStats().incMessagesBeingReceived(true, header.messageLength);
long startSer = this.getStats().startMsgDeserialization();
@@ -113,8 +112,8 @@ public class MsgReader {
void readChunk(Header header, MsgDestreamer md)
throws IOException {
- try (final ByteBufferSharing sharedBuffer = readAtLeast(header.messageLength)) {
- ByteBuffer unwrappedBuffer = sharedBuffer.getBuffer();
+ synchronized (ioFilter.getSynchObject()) {
+ ByteBuffer unwrappedBuffer = readAtLeast(header.messageLength);
this.getStats().incMessagesBeingReceived(md.size() == 0, header.messageLength);
md.addChunk(unwrappedBuffer, header.messageLength);
// show that the bytes have been consumed by adjusting the buffer's position
@@ -124,7 +123,7 @@ public class MsgReader {
- private ByteBufferSharing readAtLeast(int bytes) throws IOException {
+ private ByteBuffer readAtLeast(int bytes) throws IOException {
peerNetData = ioFilter.ensureWrappedCapacity(bytes, peerNetData,
BufferPool.BufferType.TRACKED_RECEIVER);
return ioFilter.readAtLeast(conn.getSocket().getChannel(), bytes, peerNetData);
diff --git a/geode-core/src/test/java/org/apache/geode/internal/net/ByteBufferSharingImplTest.java b/geode-core/src/test/java/org/apache/geode/internal/net/ByteBufferSharingImplTest.java
deleted file mode 100644
index bb5a75f..0000000
--- a/geode-core/src/test/java/org/apache/geode/internal/net/ByteBufferSharingImplTest.java
+++ /dev/null
@@ -1,163 +0,0 @@
-/*
- * 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.internal.net;
-
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.assertj.core.api.Assertions.fail;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-
-import java.nio.ByteBuffer;
-import java.util.concurrent.CountDownLatch;
-
-import org.junit.Before;
-import org.junit.Test;
-
-public class ByteBufferSharingImplTest {
-
- private ByteBufferSharingImpl sharing;
- private BufferPool poolMock;
- private CountDownLatch clientHasOpenedResource;
- private CountDownLatch clientMayComplete;
-
- @Before
- public void before() {
- poolMock = mock(BufferPool.class);
- sharing =
- new ByteBufferSharingImpl(mock(ByteBuffer.class), BufferPool.BufferType.TRACKED_SENDER,
- poolMock);
- clientHasOpenedResource = new CountDownLatch(1);
- clientMayComplete = new CountDownLatch(1);
- }
-
- @Test
- public void balancedCloseOwnerIsLastReferenceHolder() throws InterruptedException {
- resourceOwnerIsLastReferenceHolder("client with balanced close calls", () -> {
- try (final ByteBufferSharing _unused = sharing.open()) {
- }
- });
- }
-
- @Test
- public void extraCloseOwnerIsLastReferenceHolder() throws InterruptedException {
- resourceOwnerIsLastReferenceHolder("client with extra close calls", () -> {
- final ByteBufferSharing sharing2 = sharing.open();
- sharing2.close();
- verify(poolMock, times(0)).releaseBuffer(any(), any());
- assertThatThrownBy(() -> sharing2.close()).isInstanceOf(IllegalMonitorStateException.class);
- verify(poolMock, times(0)).releaseBuffer(any(), any());
- });
- }
-
- @Test
- public void balancedCloseClientIsLastReferenceHolder() throws InterruptedException {
- clientIsLastReferenceHolder("client with balanced close calls", () -> {
- try (final ByteBufferSharing _unused = sharing.open()) {
- clientHasOpenedResource.countDown();
- blockClient();
- }
- });
- }
-
- @Test
- public void extraCloseClientIsLastReferenceHolder() throws InterruptedException {
- clientIsLastReferenceHolder("client with extra close calls", () -> {
- final ByteBufferSharing sharing2 = sharing.open();
- clientHasOpenedResource.countDown();
- blockClient();
- sharing2.close();
- verify(poolMock, times(1)).releaseBuffer(any(), any());
- assertThatThrownBy(() -> sharing2.close()).isInstanceOf(IllegalMonitorStateException.class);
- System.out.println("here");
- });
- }
-
- @Test
- public void extraCloseDoesNotPrematurelyReturnBufferToPool() {
- final ByteBufferSharing sharing2 = sharing.open();
- sharing2.close();
- assertThatThrownBy(() -> sharing2.close()).isInstanceOf(IllegalMonitorStateException.class);
- verify(poolMock, times(0)).releaseBuffer(any(), any());
- sharing.destruct();
- verify(poolMock, times(1)).releaseBuffer(any(), any());
- }
-
- @Test
- public void extraCloseDoesNotDecrementRefCount() {
- final ByteBufferSharing sharing2 = sharing.open();
- sharing2.close();
- assertThatThrownBy(() -> sharing2.close()).isInstanceOf(IllegalMonitorStateException.class);
- final ByteBufferSharing sharing3 = this.sharing.open();
- sharing.destruct();
- verify(poolMock, times(0)).releaseBuffer(any(), any());
- }
-
- private void resourceOwnerIsLastReferenceHolder(final String name, final Runnable client)
- throws InterruptedException {
- /*
- * Thread.currentThread() is thread is playing the role of the (ByteBuffer) resource owner
- */
-
- /*
- * clientThread thread is playing the role of the client (of the resource owner)
- */
- final Thread clientThread = new Thread(client, name);
- clientThread.start();
- clientThread.join();
-
- verify(poolMock, times(0)).releaseBuffer(any(), any());
-
- sharing.destruct();
-
- verify(poolMock, times(1)).releaseBuffer(any(), any());
- }
-
- private void clientIsLastReferenceHolder(final String name, final Runnable client)
- throws InterruptedException {
- /*
- * Thread.currentThread() is thread is playing the role of the (ByteBuffer) resource owner
- */
-
- /*
- * clientThread thread is playing the role of the client (of the resource owner)
- */
- final Thread clientThread = new Thread(client, name);
- clientThread.start();
-
- clientHasOpenedResource.await();
-
- sharing.destruct();
-
- verify(poolMock, times(0)).releaseBuffer(any(), any());
-
- clientMayComplete.countDown(); // let client finish
-
- clientThread.join();
-
- verify(poolMock, times(1)).releaseBuffer(any(), any());
- }
-
- private void blockClient() {
- try {
- clientMayComplete.await();
- } catch (InterruptedException e) {
- fail("test client thread interrupted: " + e);
- }
- }
-
-}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/net/NioPlainEngineTest.java b/geode-core/src/test/java/org/apache/geode/internal/net/NioPlainEngineTest.java
index 7ab838c..3d394fb 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/net/NioPlainEngineTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/net/NioPlainEngineTest.java
@@ -50,8 +50,7 @@ public class NioPlainEngineTest {
public void unwrap() {
ByteBuffer buffer = ByteBuffer.allocate(100);
buffer.position(0).limit(buffer.capacity());
- try (final ByteBufferSharing unused = nioEngine.unwrap(buffer)) {
- }
+ nioEngine.unwrap(buffer);
assertThat(buffer.position()).isEqualTo(buffer.limit());
}
@@ -117,29 +116,23 @@ public class NioPlainEngineTest {
nioEngine.lastReadPosition = 10;
- try (final ByteBufferSharing sharedBuffer =
- nioEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer)) {
- ByteBuffer data = sharedBuffer.getBuffer();
- verify(mockChannel, times(3)).read(isA(ByteBuffer.class));
- assertThat(data.position()).isEqualTo(0);
- assertThat(data.limit()).isEqualTo(amountToRead);
- assertThat(nioEngine.lastReadPosition).isEqualTo(individualRead * 3 + preexistingBytes);
- assertThat(nioEngine.lastProcessedPosition).isEqualTo(amountToRead);
- }
-
- try (final ByteBufferSharing sharedBuffer =
- nioEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer)) {
- final ByteBuffer data = sharedBuffer.getBuffer();
- verify(mockChannel, times(5)).read(any(ByteBuffer.class));
- // at end of last readAtLeast data
- assertThat(data.position()).isEqualTo(amountToRead);
- // we read amountToRead bytes
- assertThat(data.limit()).isEqualTo(amountToRead * 2);
- // we did 2 more reads from the network
- assertThat(nioEngine.lastReadPosition).isEqualTo(individualRead * 5 + preexistingBytes);
- // the next read will start at the end of consumed data
- assertThat(nioEngine.lastProcessedPosition).isEqualTo(amountToRead * 2);
- }
+ ByteBuffer data = nioEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer);
+ verify(mockChannel, times(3)).read(isA(ByteBuffer.class));
+ assertThat(data.position()).isEqualTo(0);
+ assertThat(data.limit()).isEqualTo(amountToRead);
+ assertThat(nioEngine.lastReadPosition).isEqualTo(individualRead * 3 + preexistingBytes);
+ assertThat(nioEngine.lastProcessedPosition).isEqualTo(amountToRead);
+
+ data = nioEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer);
+ verify(mockChannel, times(5)).read(any(ByteBuffer.class));
+ // at end of last readAtLeast data
+ assertThat(data.position()).isEqualTo(amountToRead);
+ // we read amountToRead bytes
+ assertThat(data.limit()).isEqualTo(amountToRead * 2);
+ // we did 2 more reads from the network
+ assertThat(nioEngine.lastReadPosition).isEqualTo(individualRead * 5 + preexistingBytes);
+ // the next read will start at the end of consumed data
+ assertThat(nioEngine.lastProcessedPosition).isEqualTo(amountToRead * 2);
}
@@ -154,9 +147,7 @@ public class NioPlainEngineTest {
nioEngine.lastReadPosition = 10;
- try (final ByteBufferSharing unused =
- nioEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer)) {
- }
+ nioEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer);
}
}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/net/NioSslEngineTest.java b/geode-core/src/test/java/org/apache/geode/internal/net/NioSslEngineTest.java
index e9b01cf..ee4aaa3 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/net/NioSslEngineTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/net/NioSslEngineTest.java
@@ -69,7 +69,6 @@ public class NioSslEngineTest {
private DMStats mockStats;
private NioSslEngine nioSslEngine;
private NioSslEngine spyNioSslEngine;
- private BufferPool spyBufferPool;
@Before
public void setUp() throws Exception {
@@ -82,17 +81,13 @@ public class NioSslEngineTest {
mockStats = mock(DMStats.class);
- final BufferPool bufferPool = new BufferPool(mockStats);
- spyBufferPool = spy(bufferPool);
- nioSslEngine = new NioSslEngine(mockEngine, spyBufferPool);
+ nioSslEngine = new NioSslEngine(mockEngine, new BufferPool(mockStats));
spyNioSslEngine = spy(nioSslEngine);
}
@Test
- public void engineUsesDirectBuffers() throws IOException {
- try (final ByteBufferSharing outputSharing = nioSslEngine.shareOutputBuffer()) {
- assertThat(outputSharing.getBuffer().isDirect()).isTrue();
- }
+ public void engineUsesDirectBuffers() {
+ assertThat(nioSslEngine.myNetData.isDirect()).isTrue();
}
@Test
@@ -124,7 +119,7 @@ public class NioSslEngineTest {
verify(mockEngine, atLeast(2)).getHandshakeStatus();
verify(mockEngine, times(3)).wrap(any(ByteBuffer.class), any(ByteBuffer.class));
verify(mockEngine, times(3)).unwrap(any(ByteBuffer.class), any(ByteBuffer.class));
- verify(spyBufferPool, times(2)).expandWriteBufferIfNeeded(any(BufferPool.BufferType.class),
+ verify(spyNioSslEngine, times(2)).expandWriteBuffer(any(BufferPool.BufferType.class),
any(ByteBuffer.class), any(Integer.class));
verify(spyNioSslEngine, times(1)).handleBlockingTasks();
verify(mockChannel, times(3)).read(any(ByteBuffer.class));
@@ -188,148 +183,148 @@ public class NioSslEngineTest {
.hasMessageContaining("SSL Handshake terminated with status");
}
+
+ @Test
+ public void checkClosed() throws Exception {
+ nioSslEngine.checkClosed();
+ }
+
+ @Test(expected = IOException.class)
+ public void checkClosedThrows() throws Exception {
+ when(mockEngine.wrap(any(ByteBuffer.class), any(ByteBuffer.class))).thenReturn(
+ new SSLEngineResult(CLOSED, FINISHED, 0, 100));
+ nioSslEngine.close(mock(SocketChannel.class));
+ nioSslEngine.checkClosed();
+ }
+
+ @Test
+ public void synchObjectIsSelf() {
+ // for thread-safety the synchronization object given to outside entities
+ // must be the the engine itself. This allows external manipulation or
+ // use of the engine's buffers to be protected in the same way as its synchronized
+ // methods
+ assertThat(nioSslEngine.getSynchObject()).isSameAs(nioSslEngine);
+ }
+
@Test
public void wrap() throws Exception {
- try (final ByteBufferSharing outputSharing = nioSslEngine.shareOutputBuffer()) {
-
- // make the application data too big to fit into the engine's encryption buffer
- ByteBuffer appData =
- ByteBuffer.allocate(outputSharing.getBuffer().capacity() + 100);
- byte[] appBytes = new byte[appData.capacity()];
- Arrays.fill(appBytes, (byte) 0x1F);
- appData.put(appBytes);
- appData.flip();
-
- // create an engine that will transfer bytes from the application buffer to the encrypted
- // buffer
- TestSSLEngine testEngine = new TestSSLEngine();
- testEngine.addReturnResult(
- new SSLEngineResult(OK, NEED_TASK, appData.remaining(), appData.remaining()));
- spyNioSslEngine.engine = testEngine;
-
- try (final ByteBufferSharing outputSharing2 = spyNioSslEngine.wrap(appData)) {
- ByteBuffer wrappedBuffer = outputSharing2.getBuffer();
-
- verify(spyBufferPool, times(1)).expandWriteBufferIfNeeded(any(BufferPool.BufferType.class),
- any(ByteBuffer.class), any(Integer.class));
- appData.flip();
- assertThat(wrappedBuffer).isEqualTo(appData);
- }
- verify(spyNioSslEngine, times(1)).handleBlockingTasks();
- }
+ // make the application data too big to fit into the engine's encryption buffer
+ ByteBuffer appData = ByteBuffer.allocate(nioSslEngine.myNetData.capacity() + 100);
+ byte[] appBytes = new byte[appData.capacity()];
+ Arrays.fill(appBytes, (byte) 0x1F);
+ appData.put(appBytes);
+ appData.flip();
+
+ // create an engine that will transfer bytes from the application buffer to the encrypted buffer
+ TestSSLEngine testEngine = new TestSSLEngine();
+ testEngine.addReturnResult(
+ new SSLEngineResult(OK, NEED_TASK, appData.remaining(), appData.remaining()));
+ spyNioSslEngine.engine = testEngine;
+
+ ByteBuffer wrappedBuffer = spyNioSslEngine.wrap(appData);
+
+ verify(spyNioSslEngine, times(1)).expandWriteBuffer(any(BufferPool.BufferType.class),
+ any(ByteBuffer.class), any(Integer.class));
+ appData.flip();
+ assertThat(wrappedBuffer).isEqualTo(appData);
+ verify(spyNioSslEngine, times(1)).handleBlockingTasks();
}
@Test
- public void wrapFails() throws IOException {
- try (final ByteBufferSharing outputSharing = nioSslEngine.shareOutputBuffer()) {
- // make the application data too big to fit into the engine's encryption buffer
- ByteBuffer appData =
- ByteBuffer.allocate(outputSharing.getBuffer().capacity() + 100);
- byte[] appBytes = new byte[appData.capacity()];
- Arrays.fill(appBytes, (byte) 0x1F);
- appData.put(appBytes);
- appData.flip();
-
- // create an engine that will transfer bytes from the application buffer to the encrypted
- // buffer
- TestSSLEngine testEngine = new TestSSLEngine();
- testEngine.addReturnResult(
- new SSLEngineResult(CLOSED, NEED_TASK, appData.remaining(), appData.remaining()));
- spyNioSslEngine.engine = testEngine;
-
- assertThatThrownBy(() -> spyNioSslEngine.wrap(appData)).isInstanceOf(SSLException.class)
- .hasMessageContaining("Error encrypting data");
- }
+ public void wrapFails() {
+ // make the application data too big to fit into the engine's encryption buffer
+ ByteBuffer appData = ByteBuffer.allocate(nioSslEngine.myNetData.capacity() + 100);
+ byte[] appBytes = new byte[appData.capacity()];
+ Arrays.fill(appBytes, (byte) 0x1F);
+ appData.put(appBytes);
+ appData.flip();
+
+ // create an engine that will transfer bytes from the application buffer to the encrypted buffer
+ TestSSLEngine testEngine = new TestSSLEngine();
+ testEngine.addReturnResult(
+ new SSLEngineResult(CLOSED, NEED_TASK, appData.remaining(), appData.remaining()));
+ spyNioSslEngine.engine = testEngine;
+
+ assertThatThrownBy(() -> spyNioSslEngine.wrap(appData)).isInstanceOf(SSLException.class)
+ .hasMessageContaining("Error encrypting data");
}
@Test
public void unwrapWithBufferOverflow() throws Exception {
- try (final ByteBufferSharing inputSharing = nioSslEngine.shareInputBuffer()) {
- // make the application data too big to fit into the engine's encryption buffer
- final ByteBuffer peerAppData = inputSharing.getBuffer();
-
- int originalPeerAppDataCapacity = peerAppData.capacity();
- int originalPeerAppDataPosition = originalPeerAppDataCapacity / 2;
- peerAppData.position(originalPeerAppDataPosition);
- ByteBuffer wrappedData = ByteBuffer.allocate(originalPeerAppDataCapacity + 100);
- byte[] netBytes = new byte[wrappedData.capacity()];
- Arrays.fill(netBytes, (byte) 0x1F);
- wrappedData.put(netBytes);
- wrappedData.flip();
-
- // create an engine that will transfer bytes from the application buffer to the encrypted
- // buffer
- TestSSLEngine testEngine = new TestSSLEngine();
- spyNioSslEngine.engine = testEngine;
-
- testEngine.addReturnResult(
- new SSLEngineResult(BUFFER_OVERFLOW, NEED_UNWRAP, 0, 0), // results in 30,000 byte buffer
- new SSLEngineResult(BUFFER_OVERFLOW, NEED_UNWRAP, 0, 0), // 50,000 bytes
- new SSLEngineResult(BUFFER_OVERFLOW, NEED_UNWRAP, 0, 0), // 90,000 bytes
- new SSLEngineResult(OK, FINISHED, netBytes.length, netBytes.length));
-
- int expectedCapacity = 2 * originalPeerAppDataCapacity - originalPeerAppDataPosition;
- expectedCapacity =
- 2 * (expectedCapacity - originalPeerAppDataPosition) + originalPeerAppDataPosition;
- expectedCapacity =
- 2 * (expectedCapacity - originalPeerAppDataPosition) + originalPeerAppDataPosition;
- try (final ByteBufferSharing sharedBuffer = spyNioSslEngine.unwrap(wrappedData)) {
- ByteBuffer unwrappedBuffer = sharedBuffer.getBuffer();
- unwrappedBuffer.flip();
- assertThat(unwrappedBuffer.capacity()).isEqualTo(expectedCapacity);
- }
- }
+ // make the application data too big to fit into the engine's encryption buffer
+ int originalPeerAppDataCapacity = nioSslEngine.peerAppData.capacity();
+ int originalPeerAppDataPosition = originalPeerAppDataCapacity / 2;
+ nioSslEngine.peerAppData.position(originalPeerAppDataPosition);
+ ByteBuffer wrappedData = ByteBuffer.allocate(originalPeerAppDataCapacity + 100);
+ byte[] netBytes = new byte[wrappedData.capacity()];
+ Arrays.fill(netBytes, (byte) 0x1F);
+ wrappedData.put(netBytes);
+ wrappedData.flip();
+
+ // create an engine that will transfer bytes from the application buffer to the encrypted buffer
+ TestSSLEngine testEngine = new TestSSLEngine();
+ spyNioSslEngine.engine = testEngine;
+
+ testEngine.addReturnResult(
+ new SSLEngineResult(BUFFER_OVERFLOW, NEED_UNWRAP, 0, 0), // results in 30,000 byte buffer
+ new SSLEngineResult(BUFFER_OVERFLOW, NEED_UNWRAP, 0, 0), // 50,000 bytes
+ new SSLEngineResult(BUFFER_OVERFLOW, NEED_UNWRAP, 0, 0), // 90,000 bytes
+ new SSLEngineResult(OK, FINISHED, netBytes.length, netBytes.length));
+
+ int expectedCapacity = 2 * originalPeerAppDataCapacity - originalPeerAppDataPosition;
+ expectedCapacity =
+ 2 * (expectedCapacity - originalPeerAppDataPosition) + originalPeerAppDataPosition;
+ expectedCapacity =
+ 2 * (expectedCapacity - originalPeerAppDataPosition) + originalPeerAppDataPosition;
+ ByteBuffer unwrappedBuffer = spyNioSslEngine.unwrap(wrappedData);
+ unwrappedBuffer.flip();
+ assertThat(unwrappedBuffer.capacity()).isEqualTo(expectedCapacity);
}
@Test
public void unwrapWithBufferUnderflow() throws Exception {
- try (final ByteBufferSharing inputSharing = nioSslEngine.shareInputBuffer()) {
- ByteBuffer wrappedData =
- ByteBuffer.allocate(inputSharing.getBuffer().capacity());
- byte[] netBytes = new byte[wrappedData.capacity() / 2];
- Arrays.fill(netBytes, (byte) 0x1F);
- wrappedData.put(netBytes);
- wrappedData.flip();
-
- // create an engine that will transfer bytes from the application buffer to the encrypted
- // buffer
- TestSSLEngine testEngine = new TestSSLEngine();
- testEngine.addReturnResult(new SSLEngineResult(BUFFER_UNDERFLOW, NEED_TASK, 0, 0));
- spyNioSslEngine.engine = testEngine;
-
- try (final ByteBufferSharing sharedBuffer = spyNioSslEngine.unwrap(wrappedData)) {
- ByteBuffer unwrappedBuffer = sharedBuffer.getBuffer();
- unwrappedBuffer.flip();
- assertThat(unwrappedBuffer.remaining()).isEqualTo(0);
- }
- assertThat(wrappedData.position()).isEqualTo(netBytes.length);
- }
+ ByteBuffer wrappedData = ByteBuffer.allocate(nioSslEngine.peerAppData.capacity());
+ byte[] netBytes = new byte[wrappedData.capacity() / 2];
+ Arrays.fill(netBytes, (byte) 0x1F);
+ wrappedData.put(netBytes);
+ wrappedData.flip();
+
+ // create an engine that will transfer bytes from the application buffer to the encrypted buffer
+ TestSSLEngine testEngine = new TestSSLEngine();
+ testEngine.addReturnResult(new SSLEngineResult(BUFFER_UNDERFLOW, NEED_TASK, 0, 0));
+ spyNioSslEngine.engine = testEngine;
+
+ ByteBuffer unwrappedBuffer = spyNioSslEngine.unwrap(wrappedData);
+ unwrappedBuffer.flip();
+ assertThat(unwrappedBuffer.remaining()).isEqualTo(0);
+ assertThat(wrappedData.position()).isEqualTo(netBytes.length);
}
@Test
- public void unwrapWithDecryptionError() throws IOException {
- try (final ByteBufferSharing inputSharing = nioSslEngine.shareInputBuffer()) {
- // make the application data too big to fit into the engine's encryption buffer
- ByteBuffer wrappedData =
- ByteBuffer.allocate(inputSharing.getBuffer().capacity());
- byte[] netBytes = new byte[wrappedData.capacity() / 2];
- Arrays.fill(netBytes, (byte) 0x1F);
- wrappedData.put(netBytes);
- wrappedData.flip();
-
- // create an engine that will transfer bytes from the application buffer to the encrypted
- // buffer
- TestSSLEngine testEngine = new TestSSLEngine();
- testEngine.addReturnResult(new SSLEngineResult(CLOSED, FINISHED, 0, 0));
- spyNioSslEngine.engine = testEngine;
-
- assertThatThrownBy(() -> {
- try (final ByteBufferSharing unused = spyNioSslEngine.unwrap(wrappedData)) {
- }
- }).isInstanceOf(SSLException.class)
- .hasMessageContaining("Error decrypting data");
- }
+ public void unwrapWithDecryptionError() {
+ // make the application data too big to fit into the engine's encryption buffer
+ ByteBuffer wrappedData = ByteBuffer.allocate(nioSslEngine.peerAppData.capacity());
+ byte[] netBytes = new byte[wrappedData.capacity() / 2];
+ Arrays.fill(netBytes, (byte) 0x1F);
+ wrappedData.put(netBytes);
+ wrappedData.flip();
+
+ // create an engine that will transfer bytes from the application buffer to the encrypted buffer
+ TestSSLEngine testEngine = new TestSSLEngine();
+ testEngine.addReturnResult(new SSLEngineResult(CLOSED, FINISHED, 0, 0));
+ spyNioSslEngine.engine = testEngine;
+
+ assertThatThrownBy(() -> spyNioSslEngine.unwrap(wrappedData)).isInstanceOf(SSLException.class)
+ .hasMessageContaining("Error decrypting data");
+ }
+
+ @Test
+ public void ensureUnwrappedCapacity() {
+ ByteBuffer wrappedBuffer = ByteBuffer.allocate(netBufferSize);
+ int requestedCapacity = nioSslEngine.getUnwrappedBuffer(wrappedBuffer).capacity() * 2;
+ ByteBuffer unwrappedBuffer = nioSslEngine.ensureUnwrappedCapacity(requestedCapacity);
+ assertThat(unwrappedBuffer.capacity()).isGreaterThanOrEqualTo(requestedCapacity);
}
@Test
@@ -343,11 +338,7 @@ public class NioSslEngineTest {
when(mockEngine.wrap(any(ByteBuffer.class), any(ByteBuffer.class))).thenReturn(
new SSLEngineResult(CLOSED, FINISHED, 0, 0));
nioSslEngine.close(mockChannel);
- assertThatThrownBy(() -> nioSslEngine.shareOutputBuffer().getBuffer())
- .isInstanceOf(IOException.class)
- .hasMessageContaining("NioSslEngine has been closed");
- assertThatThrownBy(() -> nioSslEngine.shareInputBuffer().getBuffer())
- .isInstanceOf(IOException.class)
+ assertThatThrownBy(() -> nioSslEngine.checkClosed()).isInstanceOf(IOException.class)
.hasMessageContaining("NioSslEngine has been closed");
nioSslEngine.close(mockChannel);
}
@@ -376,12 +367,10 @@ public class NioSslEngineTest {
when(mockEngine.isOutboundDone()).thenReturn(Boolean.FALSE);
when(mockEngine.wrap(any(ByteBuffer.class), any(ByteBuffer.class))).thenAnswer((x) -> {
- try (final ByteBufferSharing outputSharing = nioSslEngine.shareOutputBuffer()) {
- // give the NioSslEngine something to write on its socket channel, simulating a TLS close
- // message
- outputSharing.getBuffer().put("Goodbye cruel world".getBytes());
- return new SSLEngineResult(CLOSED, FINISHED, 0, 0);
- }
+ // give the NioSslEngine something to write on its socket channel, simulating a TLS close
+ // message
+ nioSslEngine.myNetData.put("Goodbye cruel world".getBytes());
+ return new SSLEngineResult(CLOSED, FINISHED, 0, 0);
});
when(mockChannel.write(any(ByteBuffer.class))).thenThrow(new ClosedChannelException());
nioSslEngine.close(mockChannel);
@@ -412,42 +401,37 @@ public class NioSslEngineTest {
ByteBuffer wrappedBuffer = ByteBuffer.allocate(1000);
SocketChannel mockChannel = mock(SocketChannel.class);
- try (final ByteBufferSharing inputSharing = nioSslEngine.shareInputBuffer()) {
- // force a compaction by making the decoded buffer appear near to being full
- ByteBuffer unwrappedBuffer = inputSharing.getBuffer();
- unwrappedBuffer.position(unwrappedBuffer.capacity() - individualRead);
- unwrappedBuffer.limit(unwrappedBuffer.position() + preexistingBytes);
-
- // simulate some socket reads
- when(mockChannel.read(any(ByteBuffer.class))).thenAnswer(new Answer<Integer>() {
- @Override
- public Integer answer(InvocationOnMock invocation) throws Throwable {
- ByteBuffer buffer = invocation.getArgument(0);
- buffer.position(buffer.position() + individualRead);
- return individualRead;
- }
- });
-
- TestSSLEngine testSSLEngine = new TestSSLEngine();
- testSSLEngine.addReturnResult(new SSLEngineResult(OK, NEED_UNWRAP, 0, 0));
- nioSslEngine.engine = testSSLEngine;
-
- try (final ByteBufferSharing sharedBuffer =
- nioSslEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer)) {
- ByteBuffer data = sharedBuffer.getBuffer();
- verify(mockChannel, times(3)).read(isA(ByteBuffer.class));
- assertThat(data.position()).isEqualTo(0);
- assertThat(data.limit()).isEqualTo(individualRead * 3 + preexistingBytes);
+ // force a compaction by making the decoded buffer appear near to being full
+ ByteBuffer unwrappedBuffer = nioSslEngine.peerAppData;
+ unwrappedBuffer.position(unwrappedBuffer.capacity() - individualRead);
+ unwrappedBuffer.limit(unwrappedBuffer.position() + preexistingBytes);
+
+ // simulate some socket reads
+ when(mockChannel.read(any(ByteBuffer.class))).thenAnswer(new Answer<Integer>() {
+ @Override
+ public Integer answer(InvocationOnMock invocation) throws Throwable {
+ ByteBuffer buffer = invocation.getArgument(0);
+ buffer.position(buffer.position() + individualRead);
+ return individualRead;
}
- }
+ });
+
+ TestSSLEngine testSSLEngine = new TestSSLEngine();
+ testSSLEngine.addReturnResult(new SSLEngineResult(OK, NEED_UNWRAP, 0, 0));
+ nioSslEngine.engine = testSSLEngine;
+
+ ByteBuffer data = nioSslEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer);
+ verify(mockChannel, times(3)).read(isA(ByteBuffer.class));
+ assertThat(data.position()).isEqualTo(0);
+ assertThat(data.limit()).isEqualTo(individualRead * 3 + preexistingBytes);
}
/**
- * This tests the case where a message header has been read and part of a message has been read,
- * but the decoded buffer is too small to hold all of the message. In this case the readAtLeast
- * method will have to expand the capacity of the decoded buffer and return the new, expanded,
- * buffer as the method result.
+ * This tests the case where a message header has been read and part of a message has been
+ * read, but the decoded buffer is too small to hold all of the message. In this case
+ * the readAtLeast method will have to expand the capacity of the decoded buffer and return
+ * the new, expanded, buffer as the method result.
*/
@Test
public void readAtLeastUsingSmallAppBuffer() throws Exception {
@@ -461,11 +445,7 @@ public class NioSslEngineTest {
int initialUnwrappedBufferSize = 100;
ByteBuffer unwrappedBuffer = ByteBuffer.allocate(initialUnwrappedBufferSize);
unwrappedBuffer.position(7).limit(preexistingBytes + 7); // 7 bytes of message header - ignored
-
- try (final ByteBufferSharing inputSharing = nioSslEngine.shareInputBuffer()) {
- final ByteBufferSharingImpl inputSharingImpl = (ByteBufferSharingImpl) inputSharing;
- inputSharingImpl.setBufferForTestingOnly(unwrappedBuffer);
- }
+ nioSslEngine.peerAppData = unwrappedBuffer;
// simulate some socket reads
when(mockChannel.read(any(ByteBuffer.class))).thenAnswer(new Answer<Integer>() {
@@ -485,26 +465,22 @@ public class NioSslEngineTest {
new SSLEngineResult(OK, NEED_UNWRAP, 0, 0)); // 130 + 60 bytes = 190
nioSslEngine.engine = testSSLEngine;
- try (final ByteBufferSharing sharedBuffer =
- nioSslEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer)) {
- ByteBuffer data = sharedBuffer.getBuffer();
- verify(mockChannel, times(3)).read(isA(ByteBuffer.class));
- assertThat(data.position()).isEqualTo(0);
- assertThat(data.limit()).isEqualTo(individualRead * 3 + preexistingBytes);
- // The initial available space in the unwrapped buffer should have doubled
- int initialFreeSpace = initialUnwrappedBufferSize - preexistingBytes;
- try (final ByteBufferSharing inputSharing = nioSslEngine.shareInputBuffer()) {
- assertThat(inputSharing.getBuffer().capacity())
- .isEqualTo(2 * initialFreeSpace + preexistingBytes);
- }
- }
+ ByteBuffer data = nioSslEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer);
+ verify(mockChannel, times(3)).read(isA(ByteBuffer.class));
+ assertThat(data.position()).isEqualTo(0);
+ assertThat(data.limit()).isEqualTo(individualRead * 3 + preexistingBytes);
+ // The initial available space in the unwrapped buffer should have doubled
+ int initialFreeSpace = initialUnwrappedBufferSize - preexistingBytes;
+ assertThat(nioSslEngine.peerAppData.capacity())
+ .isEqualTo(2 * initialFreeSpace + preexistingBytes);
}
/**
- * This tests the case where a message header has been read and part of a message has been read,
- * but the decoded buffer is too small to hold all of the message. In this case the buffer is
- * completely full and should only take one overflow response to resolve the problem.
+ * This tests the case where a message header has been read and part of a message has been
+ * read, but the decoded buffer is too small to hold all of the message. In this case
+ * the buffer is completely full and should only take one overflow response to resolve
+ * the problem.
*/
@Test
public void readAtLeastUsingSmallAppBufferAtWriteLimit() throws Exception {
@@ -519,10 +495,7 @@ public class NioSslEngineTest {
// force buffer expansion by making a small decoded buffer appear near to being full
ByteBuffer unwrappedBuffer = ByteBuffer.allocate(initialUnwrappedBufferSize);
unwrappedBuffer.position(7).limit(preexistingBytes + 7); // 7 bytes of message header - ignored
- try (final ByteBufferSharing inputSharing = nioSslEngine.shareInputBuffer()) {
- final ByteBufferSharingImpl inputSharingImpl = (ByteBufferSharingImpl) inputSharing;
- inputSharingImpl.setBufferForTestingOnly(unwrappedBuffer);
- }
+ nioSslEngine.peerAppData = unwrappedBuffer;
// simulate some socket reads
when(mockChannel.read(any(ByteBuffer.class))).thenAnswer(new Answer<Integer>() {
@@ -542,14 +515,11 @@ public class NioSslEngineTest {
new SSLEngineResult(OK, NEED_UNWRAP, 0, 0));
nioSslEngine.engine = testSSLEngine;
- try (final ByteBufferSharing sharedBuffer =
- nioSslEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer)) {
- ByteBuffer data = sharedBuffer.getBuffer();
- verify(mockChannel, times(1)).read(isA(ByteBuffer.class));
- assertThat(data.position()).isEqualTo(0);
- assertThat(data.limit())
- .isEqualTo(individualRead * testSSLEngine.getNumberOfUnwraps() + preexistingBytes);
- }
+ ByteBuffer data = nioSslEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer);
+ verify(mockChannel, times(1)).read(isA(ByteBuffer.class));
+ assertThat(data.position()).isEqualTo(0);
+ assertThat(data.limit())
+ .isEqualTo(individualRead * testSSLEngine.getNumberOfUnwraps() + preexistingBytes);
}
@@ -687,8 +657,8 @@ public class NioSslEngineTest {
}
/**
- * add an engine operation result to be returned by wrap or unwrap. Like Mockito's thenReturn(),
- * the last return result will repeat forever
+ * add an engine operation result to be returned by wrap or unwrap.
+ * Like Mockito's thenReturn(), the last return result will repeat forever
*/
void addReturnResult(SSLEngineResult... sslEngineResult) {
for (SSLEngineResult result : sslEngineResult) {