DATAGEODE-344 - Implement mustGroupTransactionEvents in GatewaySenderWrapper.

Resolves gh-389.
This commit is contained in:
John Blum
2020-07-07 17:17:30 -07:00
committed by John Blum
parent 757ede0f09
commit 76bbfcab68
3 changed files with 41 additions and 24 deletions

View File

@@ -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;

View File

@@ -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
*/

View File

@@ -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;
}
}