RouterService extends AbstractService implements IRouterService { @Deprecated protected boolean producesCommonBatches(Channel channel, String nodeGroupId, List<TriggerRouter> triggerRouters) { String channelId = channel.getChannelId(); Boolean producesCommonBatches = commonBatchesLastKnownState.get(channelId); long cacheTime = parameterService.getLong(ParameterConstants.CACHE_CHANNEL_COMMON_BATCHES_IN_MS); if (producesCommonBatches == null || System.currentTimeMillis() - commonBatchesCacheTime > cacheTime) { producesCommonBatches = !Constants.CHANNEL_CONFIG.equals(channelId) && !channel.isFileSyncFlag() && !channel.isReloadFlag() && !Constants.CHANNEL_HEARTBEAT.equals(channelId) && !Constants.CHANNEL_MONITOR.equals(channelId); if (producesCommonBatches && triggerRouters != null) { List<TriggerRouter> testableTriggerRouters = new ArrayList<TriggerRouter>(); for (TriggerRouter triggerRouter : triggerRouters) { if (triggerRouter.getTrigger().getChannelId().equals(channelId)) { testableTriggerRouters.add(triggerRouter); } else { String anotherChannelTableName = triggerRouter.getTrigger() .getFullyQualifiedSourceTableName(); for (TriggerRouter triggerRouter2 : triggerRouters) { String currentTableName = triggerRouter2 .getTrigger() .getFullyQualifiedSourceTableName(); String currentChannelId = triggerRouter2.getTrigger().getChannelId(); if (anotherChannelTableName.equals(currentTableName) && currentChannelId.equals(channelId) && triggerRouter.getRouter().getNodeGroupLink().getTargetNodeGroupId() .equals(triggerRouter2.getRouter().getNodeGroupLink().getSourceNodeGroupId()) && triggerRouter.getRouter().getNodeGroupLink().getSourceNodeGroupId() .equals(triggerRouter2.getRouter().getNodeGroupLink().getTargetNodeGroupId())) { testableTriggerRouters.add(triggerRouter); } } } } for (TriggerRouter triggerRouter : testableTriggerRouters) { boolean isDefaultRouter = "default".equals(triggerRouter.getRouter().getRouterType()); if (triggerRouter.getRouter().getNodeGroupLink().getSourceNodeGroupId() .equals(nodeGroupId)) { if (!isDefaultRouter) { producesCommonBatches = false; break; } else { if (triggerRouter.getTrigger().isSyncOnIncomingBatch()) { String outgoingTableName = triggerRouter.getTrigger() .getFullyQualifiedSourceTableName(); for (TriggerRouter triggerRouter2 : testableTriggerRouters) { String incomingTableName = triggerRouter2.getTrigger().getFullyQualifiedSourceTableName(); String targetNodeGroupId = triggerRouter2.getRouter().getNodeGroupLink() .getTargetNodeGroupId(); if (incomingTableName.equals(outgoingTableName) && targetNodeGroupId.equals(nodeGroupId)) { producesCommonBatches = false; break; } } } } } } } if (!producesCommonBatches.equals(commonBatchesLastKnownState.get(channelId))) { String message = "The '{}' channel is " + (producesCommonBatches ? "" : "NOT ") + "in common batch mode"; if (channelId.equals(Constants.CHANNEL_CONFIG) || channelId.equals(Constants.CHANNEL_HEARTBEAT) || channelId.equals(Constants.CHANNEL_FILESYNC) || channelId.equals(Constants.CHANNEL_MONITOR) || channelId.equals(Constants.CHANNEL_FILESYNC_RELOAD) || channelId.equals(Constants.CHANNEL_RELOAD)) { log.debug(message, channelId); } else { log.info(message, channelId); } commonBatchesLastKnownState.put(channelId, producesCommonBatches); } commonBatchesCacheTime = System.currentTimeMillis(); } return producesCommonBatches; } RouterService(ISymmetricEngine engine); boolean shouldDataBeRouted(SimpleRouterContext context, DataMetaData dataMetaData,
            Node node, boolean initialLoad, boolean initialLoadSelectUsed, TriggerRouter triggerRouter); synchronized void stop(); synchronized long routeData(boolean force); long getUnroutedDataCount(); List<String> getAvailableBatchAlgorithms(); Map<String, IDataRouter> getRouters(); List<DataGap> getDataGaps();  }
@Test public void testProducesCommonBatchesOneTableOneChannelDefaultRouter() { List<TriggerRouter> triggerRouters = new ArrayList<TriggerRouter>(); triggerRouters.add(new TriggerRouter(new Trigger("a", CHANNEL_2_TEST.getChannelId()), new Router("test", SOURCE_NODE_GROUP, TARGET_NODE_GROUP, "default"))); assertTrue(routerService.producesCommonBatches(CHANNEL_2_TEST, SOURCE_NODE_GROUP, triggerRouters)); }
@Test public void testNotProducesCommonBatchesOneTableOneChannelNonDefaultRouter() { List<TriggerRouter> triggerRouters = new ArrayList<TriggerRouter>(); triggerRouters.add(new TriggerRouter(new Trigger("a", CHANNEL_2_TEST.getChannelId()), new Router("test", SOURCE_NODE_GROUP, TARGET_NODE_GROUP, "column"))); assertTrue(!routerService.producesCommonBatches(CHANNEL_2_TEST, SOURCE_NODE_GROUP, triggerRouters)); }
@Test public void testProducesCommonBatchesMultipleTablesTwoChannelsMultipleRouters() { List<TriggerRouter> triggerRouters = new ArrayList<TriggerRouter>(); triggerRouters.add(new TriggerRouter(new Trigger("a", CHANNEL_2_TEST.getChannelId()), new Router("test1", SOURCE_NODE_GROUP, TARGET_NODE_GROUP, "default"))); triggerRouters.add(new TriggerRouter(new Trigger("b", "anotherchannel"), new Router("test2", SOURCE_NODE_GROUP, TARGET_NODE_GROUP, "column"))); assertTrue(routerService.producesCommonBatches(CHANNEL_2_TEST, SOURCE_NODE_GROUP, triggerRouters)); }
@Test public void testProducesCommonBatchesMultipleTablesTwoChannelsMultipleRoutersBidirectional() { List<TriggerRouter> triggerRouters = new ArrayList<TriggerRouter>(); triggerRouters.add(new TriggerRouter(new Trigger("a", CHANNEL_2_TEST.getChannelId()), new Router("test", SOURCE_NODE_GROUP, TARGET_NODE_GROUP, "default"))); triggerRouters.add(new TriggerRouter(new Trigger("a", CHANNEL_2_TEST.getChannelId()), new Router("test", TARGET_NODE_GROUP, SOURCE_NODE_GROUP, "default"))); assertTrue(routerService.producesCommonBatches(CHANNEL_2_TEST, SOURCE_NODE_GROUP, triggerRouters)); }
@Test public void testNotProducesCommonBatchesMultipleTablesTwoChannelsMultipleRoutersSyncOnIncoming() { List<TriggerRouter> triggerRouters = new ArrayList<TriggerRouter>(); Trigger tableTrigger = new Trigger("a", CHANNEL_2_TEST.getChannelId(), true); triggerRouters.add(new TriggerRouter(tableTrigger, new Router("test", SOURCE_NODE_GROUP, TARGET_NODE_GROUP, "default"))); triggerRouters.add(new TriggerRouter(tableTrigger, new Router("test", TARGET_NODE_GROUP, SOURCE_NODE_GROUP, "default"))); assertTrue(!routerService.producesCommonBatches(CHANNEL_2_TEST, SOURCE_NODE_GROUP, triggerRouters)); }
@Test public void testNotProducesCommonBatchesSameTablesTwoChannelsMultipleRoutersSameTableIncomingOnAnotherChannel() { List<TriggerRouter> triggerRouters = new ArrayList<TriggerRouter>(); Trigger tableTrigger1 = new Trigger("a", CHANNEL_2_TEST.getChannelId(), true); Trigger tableTrigger2= new Trigger("a", "anotherchannel"); triggerRouters.add(new TriggerRouter(tableTrigger1, new Router("test", SOURCE_NODE_GROUP, TARGET_NODE_GROUP, "default"))); triggerRouters.add(new TriggerRouter(tableTrigger2, new Router("test", TARGET_NODE_GROUP, SOURCE_NODE_GROUP, "default"))); assertTrue(!routerService.producesCommonBatches(CHANNEL_2_TEST, SOURCE_NODE_GROUP, triggerRouters)); }
@Test public void testProducesCommonBatchesSameTablesTwoChannelsMultipleRoutersDifferentTableIncomingOnAnotherChannel() { List<TriggerRouter> triggerRouters = new ArrayList<TriggerRouter>(); Trigger tableTrigger1 = new Trigger("a", CHANNEL_2_TEST.getChannelId(), true); Trigger tableTrigger2= new Trigger("b", "anotherchannel"); Trigger tableTrigger3= new Trigger("c", CHANNEL_2_TEST.getChannelId()); triggerRouters.add(new TriggerRouter(tableTrigger1, new Router("test", SOURCE_NODE_GROUP, TARGET_NODE_GROUP, "default"))); triggerRouters.add(new TriggerRouter(tableTrigger2, new Router("test", TARGET_NODE_GROUP, SOURCE_NODE_GROUP, "default"))); triggerRouters.add(new TriggerRouter(tableTrigger3, new Router("test", TARGET_NODE_GROUP, SOURCE_NODE_GROUP, "default"))); assertTrue(routerService.producesCommonBatches(CHANNEL_2_TEST, SOURCE_NODE_GROUP, triggerRouters)); }