From 76bbfcab68d017bdfb375c2e8c0d45dce5d1644e Mon Sep 17 00:00:00 2001
From: John Blum <jblum@pivotal.io>
Date: Tue, 7 Jul 2020 17:17:30 -0700
Subject: [PATCH] DATAGEODE-344 - Implement mustGroupTransactionEvents in
 GatewaySenderWrapper.

Resolves gh-389.
---
 .../gemfire/wan/GatewaySenderFactoryBean.java | 19 ++++++++++--
 .../gemfire/wan/GatewaySenderWrapper.java     | 15 ++++++---
 .../test/StubGatewaySenderFactory.java        | 31 +++++++++----------
 3 files changed, 41 insertions(+), 24 deletions(-)

diff --git a/spring-data-geode/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java b/spring-data-geode/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java
index f4b5c6fc..fd02bc8c 100644
--- a/spring-data-geode/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java
+++ b/spring-data-geode/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderFactoryBean.java
@@ -36,6 +36,8 @@ import org.springframework.beans.factory.FactoryBean;
 import org.springframework.data.gemfire.config.annotation.GatewaySenderConfigurer;
 import org.springframework.data.gemfire.util.ArrayUtils;
 import org.springframework.data.gemfire.util.CollectionUtils;
+import org.springframework.lang.NonNull;
+import org.springframework.lang.Nullable;
 import org.springframework.util.StringUtils;
 
 /**
@@ -72,6 +74,7 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
 
 	private Boolean diskSynchronous;
 	private Boolean batchConflationEnabled;
+	private Boolean groupTransactionEvents;
 	private Boolean parallel;
 	private Boolean persistent;
 
@@ -136,6 +139,8 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
 		Optional.ofNullable(getEventSubstitutionFilter())
 			.ifPresent(gatewaySenderFactory::setGatewayEventSubstitutionFilter);
 
+		Optional.ofNullable(getGroupTransactionEvents()).ifPresent(gatewaySenderFactory::setGroupTransactionEvents);
+
 		gatewaySenderFactory.setManualStart(isManualStart());
 
 		Optional.ofNullable(getMaximumQueueMemory()).ifPresent(gatewaySenderFactory::setMaximumQueueMemory);
@@ -182,15 +187,15 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
 			: GatewaySender.class;
 	}
 
-	public void setGatewaySender(GatewaySender gatewaySender) {
+	public void setGatewaySender(@Nullable GatewaySender gatewaySender) {
 		this.gatewaySender = gatewaySender;
 	}
 
-	public GatewaySender getGatewaySender() {
+	public @Nullable GatewaySender getGatewaySender() {
 		return this.gatewaySender;
 	}
 
-	public void setGatewaySenderConfigurers(List<GatewaySenderConfigurer> gatewaySenderConfigurers) {
+	public void setGatewaySenderConfigurers(@NonNull List<GatewaySenderConfigurer> gatewaySenderConfigurers) {
 		this.gatewaySenderConfigurers = gatewaySenderConfigurers;
 	}
 
@@ -272,6 +277,14 @@ public class GatewaySenderFactoryBean extends AbstractWANComponentFactoryBean<Ga
 		return this.eventSubstitutionFilter;
 	}
 
+	public void setGroupTransactionEvents(Boolean groupTransactionEvents) {
+		this.groupTransactionEvents = groupTransactionEvents;
+	}
+
+	public Boolean getGroupTransactionEvents() {
+		return this.groupTransactionEvents;
+	}
+
 	@Deprecated
 	public void setManualStart(boolean manualStart) {
 		this.manualStart = manualStart;
diff --git a/spring-data-geode/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderWrapper.java b/spring-data-geode/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderWrapper.java
index 45f609e1..9d8bdebb 100644
--- a/spring-data-geode/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderWrapper.java
+++ b/spring-data-geode/src/main/java/org/springframework/data/gemfire/wan/GatewaySenderWrapper.java
@@ -10,7 +10,6 @@
  * 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.springframework.data.gemfire.wan;
 
 import java.util.List;
@@ -23,12 +22,11 @@ import org.apache.geode.cache.wan.GatewayTransportFilter;
 import org.springframework.util.Assert;
 
 /**
- * {@link GatewaySenderWrapper} is an Adapter around a {@link GatewaySender} providing the ability to control
- * manual start and stop.
+ * {@link GatewaySenderWrapper} is an {@literal Adapter} around an Apache Geode {@link GatewaySender}
+ * providing the ability to control manual start and stop of the sender.
  *
  * @author David Turanski
  * @author John Blum
- * @see org.apache.geode.cache.util.Gateway
  * @see org.apache.geode.cache.wan.GatewaySender
  */
 public class GatewaySenderWrapper implements GatewaySender {
@@ -201,7 +199,6 @@ public class GatewaySenderWrapper implements GatewaySender {
 	 * @inheritDoc
 	 */
 	@Override
-	@SuppressWarnings("deprecation")
 	public GatewaySender.OrderPolicy getOrderPolicy() {
 		return delegate.getOrderPolicy();
 	}
@@ -238,6 +235,14 @@ public class GatewaySenderWrapper implements GatewaySender {
 		delegate.addGatewayEventFilter(filter);
 	}
 
+	/**
+	 * @inheritDoc
+	 */
+	@Override
+	public boolean mustGroupTransactionEvents() {
+		return this.delegate.mustGroupTransactionEvents();
+	}
+
 	/**
 	 * @inheritDoc
 	 */
diff --git a/spring-data-geode/src/test/java/org/springframework/data/gemfire/test/StubGatewaySenderFactory.java b/spring-data-geode/src/test/java/org/springframework/data/gemfire/test/StubGatewaySenderFactory.java
index 57ac1b9b..3edb6282 100644
--- a/spring-data-geode/src/test/java/org/springframework/data/gemfire/test/StubGatewaySenderFactory.java
+++ b/spring-data-geode/src/test/java/org/springframework/data/gemfire/test/StubGatewaySenderFactory.java
@@ -10,7 +10,6 @@
  * 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.springframework.data.gemfire.test;
 
 import static org.mockito.Mockito.doAnswer;
@@ -20,7 +19,6 @@ import static org.mockito.Mockito.when;
 import java.util.ArrayList;
 import java.util.List;
 
-import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
 import org.apache.geode.cache.wan.GatewayEventFilter;
@@ -31,7 +29,7 @@ import org.apache.geode.cache.wan.GatewaySenderFactory;
 import org.apache.geode.cache.wan.GatewayTransportFilter;
 
 /**
- * The StubGatewaySenderFactory class for testing purposes.
+ * {@link StubGatewaySenderFactory} class for testing purposes.
  *
  * @author David Turanski
  * @author John Blum
@@ -42,6 +40,7 @@ public class StubGatewaySenderFactory implements GatewaySenderFactory {
 
 	private boolean batchConflationEnabled;
 	private boolean diskSynchronous;
+	private boolean groupTransactionEvents;
 	private boolean manualStart;
 	private boolean parallel;
 	private boolean persistenceEnabled;
@@ -83,7 +82,8 @@ public class StubGatewaySenderFactory implements GatewaySenderFactory {
 	}
 
 	@Override
-	public GatewaySender create(final String name, final int remoteSystemId) {
+	public GatewaySender create(String name, int remoteSystemId) {
+
 		GatewaySender gatewaySender = mock(GatewaySender.class);
 
 		when(gatewaySender.getId()).thenReturn(name);
@@ -105,17 +105,11 @@ public class StubGatewaySenderFactory implements GatewaySenderFactory {
 		when(gatewaySender.isParallel()).thenReturn(this.parallel);
 		when(gatewaySender.isPersistenceEnabled()).thenReturn(this.persistenceEnabled);
 		when(gatewaySender.getOrderPolicy()).thenReturn(this.orderPolicy);
-		when(gatewaySender.isRunning()).thenAnswer(new Answer<Boolean>() {
-			@Override
-			public Boolean answer(InvocationOnMock invocation) throws Throwable {
-				return running;
-			}
-		});
-		doAnswer(new Answer<Void>() {
-			public Void answer(InvocationOnMock invocation) {
-				running = true;
-				return null;
-			}
+		when(gatewaySender.mustGroupTransactionEvents()).thenReturn(this.groupTransactionEvents);
+		when(gatewaySender.isRunning()).thenAnswer((Answer<Boolean>) invocation -> this.running);
+		doAnswer(invocation -> {
+			running = true;
+			return null;
 		}).when(gatewaySender).start();
 
 		return gatewaySender;
@@ -169,6 +163,12 @@ public class StubGatewaySenderFactory implements GatewaySenderFactory {
 		return this;
 	}
 
+	@Override
+	public GatewaySenderFactory setGroupTransactionEvents(boolean groupTransactionEvents) {
+		this.groupTransactionEvents = groupTransactionEvents;
+		return this;
+	}
+
 	@Override
 	public GatewaySenderFactory setManualStart(boolean manualStart) {
 		this.manualStart = manualStart;
@@ -228,5 +228,4 @@ public class StubGatewaySenderFactory implements GatewaySenderFactory {
 		this.gatewayEventSubstitutionFilter = gatewayEventSubstitutionFilter;
 		return this;
 	}
-
 }