CreateMappingFunction extends CliFunction<Object[]> { @Override public CliFunctionResult executeFunction(FunctionContext<Object[]> context) throws Exception { JdbcConnectorService service = FunctionContextArgumentProvider.getJdbcConnectorService(context); Object[] arguments = context.getArguments(); RegionMapping regionMapping = (RegionMapping) arguments[0]; boolean synchronous = (boolean) arguments[1]; String regionName = regionMapping.getRegionName(); Region<?, ?> region = verifyRegionExists(context.getCache(), regionName); String member = context.getMemberName(); String queueName = MappingCommandUtils.createAsyncEventQueueName(regionName); alterRegion(region, queueName, synchronous); if (isAccessor(region)) { return new CliFunctionResult(member, true, MappingConstants.THERE_IS_NO_JDBC_MAPPING_ON_PROXY_REGION); } else if (!synchronous) { createAsyncEventQueue(context.getCache(), queueName, region.getAttributes().getDataPolicy().withPartitioning()); } createRegionMapping(service, regionMapping); String message = "Created JDBC mapping for region " + regionMapping.getRegionName() + " on " + member; return new CliFunctionResult(member, true, message); } CreateMappingFunction(); @Override CliFunctionResult executeFunction(FunctionContext<Object[]> context);  }
@Test public void createRegionMappingThrowsIfRegionDoesNotExist() { when(cache.getRegion(REGION_NAME)).thenReturn(null); Throwable throwable = catchThrowable(() -> function.executeFunction(context)); assertThat(throwable).isInstanceOf(IllegalStateException.class) .hasMessage("create jdbc-mapping requires that the region \"" + REGION_NAME + "\" exists."); }
@Test public void executeWithSynchronousAltersRegionWriter() throws Exception { setupSynchronous(); function.executeFunction(context); AttributesMutator<Object, Object> mutator = region.getAttributesMutator(); verify(mutator, times(1)).setCacheWriter(any()); }
@Test public void executeWithSynchronousNeverAltersRegionAsyncEventQueue() throws Exception { setupSynchronous(); function.executeFunction(context); AttributesMutator<Object, Object> mutator = region.getAttributesMutator(); verify(mutator, never()).addAsyncEventQueueId(any()); }
@Test public void executeWithSynchronousNeverCreatesAsyncQueue() throws Exception { setupSynchronous(); function.executeFunction(context); verify(asyncEventQueueFactory, never()).create(any(), any()); }
@Test public void executeCreatesSerialAsyncQueueForNonPartitionedRegion() throws Exception { when(attributes.getDataPolicy()).thenReturn(DataPolicy.REPLICATE); function.executeFunction(context); verify(asyncEventQueueFactory, times(1)).create(any(), any()); verify(asyncEventQueueFactory, times(1)).setParallel(false); }
@SuppressWarnings("unchecked") @Test public void shouldOnlyAddAEQIdForProxyPR() throws Exception { when(attributes.getDataPolicy()).thenReturn(DataPolicy.PARTITION); PartitionedRegion pr = mock(PartitionedRegion.class); when(cache.getRegion(REGION_NAME)).thenReturn(pr); when(pr.getAttributes()).thenReturn(attributes); PartitionAttributes<Object, Object> pa = mock(PartitionAttributes.class); when(attributes.getPartitionAttributes()).thenReturn(pa); when(pa.getLocalMaxMemory()).thenReturn(0); when(pr.getAttributesMutator()).thenReturn(mock(AttributesMutator.class)); function.executeFunction(context); @SuppressWarnings("unchecked") AttributesMutator<Object, Object> mutator = pr.getAttributesMutator(); verify(mutator, times(1)).addAsyncEventQueueId(any()); verify(mutator, times(0)).setCacheWriter(any()); verify(mutator, times(0)).setCacheLoader(any()); verify(asyncEventQueueFactory, times(0)).create(any(), any()); }
@Test public void shouldOnlyAddAEQIdForProxyReplicate() throws Exception { when(attributes.getDataPolicy()).thenReturn(DataPolicy.EMPTY); when(cache.getRegion(REGION_NAME)).thenReturn(region); LocalRegion lr = (LocalRegion) region; when(lr.getAttributes()).thenReturn(attributes); when(lr.getAttributesMutator()).thenReturn(mock(AttributesMutator.class)); function.executeFunction(context); @SuppressWarnings("unchecked") AttributesMutator<Object, Object> mutator = lr.getAttributesMutator(); verify(mutator, times(1)).addAsyncEventQueueId(any()); verify(mutator, times(0)).setCacheWriter(any()); verify(mutator, times(0)).setCacheLoader(any()); verify(asyncEventQueueFactory, times(0)).create(any(), any()); }
@Test public void executeCreatesParallelAsyncQueueForPartitionedRegion() throws Exception { when(attributes.getDataPolicy()).thenReturn(DataPolicy.PARTITION); function.executeFunction(context); verify(asyncEventQueueFactory, times(1)).create(any(), any()); verify(asyncEventQueueFactory, times(1)).setParallel(true); }
@Test public void executeCreatesPartitionedProxyShouldNotContainWriterOrLoader() throws Exception { when(attributes.getDataPolicy()).thenReturn(DataPolicy.PARTITION); function.executeFunction(context); verify(asyncEventQueueFactory, times(1)).create(any(), any()); verify(asyncEventQueueFactory, times(1)).setParallel(true); }