ConfigurationChangedDataRouter extends AbstractDataRouter implements IDataRouter, IBuiltInExtensionPoint { public Set<String> routeToNodes(SimpleRouterContext routingContext, DataMetaData dataMetaData, Set<Node> possibleTargetNodes, boolean initialLoad, boolean initialLoadSelectUsed, TriggerRouter triggerRouter) { possibleTargetNodes = filterOutOlderNodes(dataMetaData, possibleTargetNodes); possibleTargetNodes = filterOutNodesByDeploymentType(dataMetaData, possibleTargetNodes); Set<String> nodeIds = new HashSet<String>(); Map<String, String> columnValues = getDataMap(dataMetaData, engine != null ? engine.getSymmetricDialect() : null); Node me = findIdentity(); if (me != null) { NetworkedNode rootNetworkedNode = getRootNetworkNodeFromContext(routingContext); if (tableMatches(dataMetaData, TableConstants.SYM_NODE) && dataMetaData.getData().getDataEventType().equals(DataEventType.SQL) && dataMetaData.getData().getParsedData(CsvData.ROW_DATA).length > 1 && dataMetaData.getData().getParsedData(CsvData.ROW_DATA)[0].toUpperCase().contains("TABLE")) { routingContext.put(CTX_KEY_RESYNC_NEEDED, Boolean.TRUE); routeNodeTables(nodeIds, columnValues, rootNetworkedNode, me, routingContext, dataMetaData, possibleTargetNodes, initialLoad); } else if (tableMatches(dataMetaData, TableConstants.SYM_NODE) || tableMatches(dataMetaData, TableConstants.SYM_NODE_SECURITY) || tableMatches(dataMetaData, TableConstants.SYM_NODE_HOST) || tableMatches(dataMetaData, TableConstants.SYM_MONITOR_EVENT)) { if (tableMatches(dataMetaData, TableConstants.SYM_NODE)) { routingContext.put(CTX_KEY_FLUSH_NODES_NEEDED, Boolean.TRUE); } else if (tableMatches(dataMetaData, TableConstants.SYM_NODE_SECURITY)) { routingContext.put(CTX_KEY_FLUSH_NODE_SECURITYS_NEEDED, Boolean.TRUE); } routeNodeTables(nodeIds, columnValues, rootNetworkedNode, me, routingContext, dataMetaData, possibleTargetNodes, initialLoad); } else if (tableMatches(dataMetaData, TableConstants.SYM_TABLE_RELOAD_REQUEST)) { String sourceNodeId = columnValues.get("SOURCE_NODE_ID"); String targetNodeId = columnValues.get("TARGET_NODE_ID"); for (Node nodeThatMayBeRoutedTo : possibleTargetNodes) { if (!Constants.DEPLOYMENT_TYPE_REST.equals(nodeThatMayBeRoutedTo .getDeploymentType()) && !nodeThatMayBeRoutedTo.requires13Compatiblity() && (nodeThatMayBeRoutedTo.getNodeId().equals(sourceNodeId) || nodeThatMayBeRoutedTo.getNodeId().equals(targetNodeId))) { nodeIds.add(nodeThatMayBeRoutedTo.getNodeId()); } } } else { IConfigurationService configurationService = engine.getConfigurationService(); for (Node nodeThatMayBeRoutedTo : possibleTargetNodes) { if (!Constants.DEPLOYMENT_TYPE_REST.equals(nodeThatMayBeRoutedTo .getDeploymentType()) && !nodeThatMayBeRoutedTo.requires13Compatiblity() && (initialLoad || !isSameNumberOfLinksAwayFromRoot(nodeThatMayBeRoutedTo, rootNetworkedNode, me))) { NodeGroupLink link = configurationService.getNodeGroupLinkFor( me.getNodeGroupId(), nodeThatMayBeRoutedTo.getNodeGroupId(), false); if (initialLoad || (link != null && link.isSyncConfigEnabled())) { nodeIds.add(nodeThatMayBeRoutedTo.getNodeId()); } } } if (StringUtils.isBlank(dataMetaData.getData().getSourceNodeId())) { queueSyncTriggers(routingContext, dataMetaData, columnValues); } if (tableMatches(dataMetaData, TableConstants.SYM_CHANNEL)) { routingContext.put(CTX_KEY_FLUSH_CHANNELS_NEEDED, Boolean.TRUE); } if (tableMatches(dataMetaData, TableConstants.SYM_CONFLICT)) { routingContext.put(CTX_KEY_FLUSH_CONFLICTS_NEEDED, Boolean.TRUE); } if (tableMatches(dataMetaData, TableConstants.SYM_LOAD_FILTER)) { routingContext.put(CTX_KEY_FLUSH_LOADFILTERS_NEEDED, Boolean.TRUE); } if (tableMatches(dataMetaData, TableConstants.SYM_PARAMETER)) { routingContext.put(CTX_KEY_FLUSH_PARAMETERS_NEEDED, Boolean.TRUE); if (StringUtils.isBlank(dataMetaData.getData().getSourceNodeId()) && (dataMetaData.getData().getRowData() != null && dataMetaData .getData().getRowData().contains("job."))) { routingContext.put(CTX_KEY_RESTART_JOBMANAGER_NEEDED, Boolean.TRUE); } } if (tableMatches(dataMetaData, TableConstants.SYM_TRANSFORM_COLUMN) || tableMatches(dataMetaData, TableConstants.SYM_TRANSFORM_TABLE)) { routingContext.put(CTX_KEY_FLUSH_TRANSFORMS_NEEDED, Boolean.TRUE); } if (tableMatches(dataMetaData, TableConstants.SYM_EXTENSION)) { routingContext.put(CTX_KEY_REFRESH_EXTENSIONS_NEEDED, Boolean.TRUE); } if (tableMatches(dataMetaData, TableConstants.SYM_MONITOR)) { routingContext.put(CTX_KEY_FLUSH_MONITORS_NEEDED, Boolean.TRUE); } if (tableMatches(dataMetaData, TableConstants.SYM_NOTIFICATION)) { routingContext.put(CTX_KEY_FLUSH_NOTIFICATIONS_NEEDED, Boolean.TRUE); } if (tableMatches(dataMetaData, TableConstants.SYM_NODE_GROUP_LINK)) { routingContext.put(CTX_KEY_FLUSH_NODE_GROUP_LINK_NEEDED, Boolean.TRUE); if(dataMetaData.getData().getDataEventType().equals(DataEventType.INSERT)) { if(! initialLoad) { if (! isConfigDataMetaDataAlreadyHandled(dataMetaData, routingContext)) { buildReloadEvents(dataMetaData, columnValues); addConfigDataMetaData(dataMetaData, routingContext); } } } } if (tableMatches(dataMetaData, TableConstants.SYM_JOB) || routingContext.get(CTX_KEY_FILE_SYNC_TRIGGERS_NEEDED) != null) { routingContext.put(CTX_KEY_FLUSH_JOBS_NEEDED, Boolean.TRUE); } } } return nodeIds; } ConfigurationChangedDataRouter();  ConfigurationChangedDataRouter(ISymmetricEngine engine); Set<String> routeToNodes(SimpleRouterContext routingContext, DataMetaData dataMetaData,
            Set<Node> possibleTargetNodes, boolean initialLoad, boolean initialLoadSelectUsed,
            TriggerRouter triggerRouter); @Override void contextCommitted(SimpleRouterContext routingContext); @Override boolean isConfigurable(); static final String ROUTER_TYPE; final static String KEY; }
@Test public void testRouteHeartbeatToParent() { IDataRouter router = buildTestableRouter( THREE_TIER_NETWORKED_ROOT.findNetworkedNode("laptop1").getNode(), THREE_TIER_LINKS, THREE_TIER_NETWORKED_ROOT); Set<Node> nodes = new HashSet<Node>(); nodes.add(THREE_TIER_NETWORKED_ROOT.findNetworkedNode("rgn1").getNode()); Collection<String> nodeIds = router.routeToNodes(new SimpleRouterContext(), buildDataMetaData("SYM_NODE", "laptop1"), nodes, false, false, null); assertNotNull(nodeIds); assertEquals(1, nodeIds.size()); assertEquals("rgn1", nodeIds.iterator().next()); }
@Test public void testRouteLaptop1FromRgn1() { IDataRouter router = buildTestableRouter( THREE_TIER_NETWORKED_ROOT.findNetworkedNode("corp").getNode(), THREE_TIER_LINKS, THREE_TIER_NETWORKED_ROOT); Set<Node> nodes = new HashSet<Node>(); nodes.add(THREE_TIER_NETWORKED_ROOT.findNetworkedNode("laptop1").getNode()); Collection<String> nodeIds = router.routeToNodes(new SimpleRouterContext(), buildDataMetaData("SYM_NODE", "laptop1"), nodes, false, false, null); assertNotNull(nodeIds); assertEquals(1, nodeIds.size()); assertEquals("laptop1", nodeIds.iterator().next()); }
@Test public void testRouteRgn2FromCorp() { IDataRouter router = buildTestableRouter( THREE_TIER_NETWORKED_ROOT.findNetworkedNode("corp").getNode(), THREE_TIER_LINKS, THREE_TIER_NETWORKED_ROOT); Set<Node> nodes = new HashSet<Node>(); nodes.add(THREE_TIER_NETWORKED_ROOT.findNetworkedNode("rgn1").getNode()); nodes.add(THREE_TIER_NETWORKED_ROOT.findNetworkedNode("rgn2").getNode()); Collection<String> nodeIds = router.routeToNodes(new SimpleRouterContext(), buildDataMetaData("SYM_NODE", "rgn2"), nodes, false, false, null); assertNotNull(nodeIds); assertEquals(1, nodeIds.size()); assertEquals("rgn2", nodeIds.iterator().next()); }
@Test public void testConfigurationExtract() { IDataRouter router = buildTestableRouter( THREE_TIER_NETWORKED_ROOT.findNetworkedNode("corp").getNode(), THREE_TIER_LINKS, THREE_TIER_NETWORKED_ROOT); Set<Node> nodes = new HashSet<Node>(); nodes.add(THREE_TIER_NETWORKED_ROOT.findNetworkedNode("corp").getNode()); Collection<String> nodeIds = router.routeToNodes(new SimpleRouterContext(), buildDataMetaData("SYM_NODE", "corp"), nodes, true, false, null); assertNotNull(nodeIds); assertEquals(1, nodeIds.size()); assertEquals("corp", nodeIds.iterator().next()); }
@Test public void testRouteRgn1FromCorp() { IDataRouter router = buildTestableRouter( THREE_TIER_NETWORKED_ROOT.findNetworkedNode("corp").getNode(), THREE_TIER_LINKS, THREE_TIER_NETWORKED_ROOT); Set<Node> nodes = new HashSet<Node>(); nodes.add(THREE_TIER_NETWORKED_ROOT.findNetworkedNode("rgn1").getNode()); nodes.add(THREE_TIER_NETWORKED_ROOT.findNetworkedNode("rgn2").getNode()); Collection<String> nodeIds = router.routeToNodes(new SimpleRouterContext(), buildDataMetaData("SYM_NODE", "rgn1"), nodes, false, false, null); assertNotNull(nodeIds); assertEquals(1, nodeIds.size()); assertEquals("rgn1", nodeIds.iterator().next()); }
@Test public void testRouteLaptop1FromCorp() { IDataRouter router = buildTestableRouter( THREE_TIER_NETWORKED_ROOT.findNetworkedNode("corp").getNode(), THREE_TIER_LINKS, THREE_TIER_NETWORKED_ROOT); Set<Node> nodes = new HashSet<Node>(); nodes.add(THREE_TIER_NETWORKED_ROOT.findNetworkedNode("rgn1").getNode()); nodes.add(THREE_TIER_NETWORKED_ROOT.findNetworkedNode("rgn2").getNode()); Collection<String> nodeIds = router.routeToNodes(new SimpleRouterContext(), buildDataMetaData("SYM_NODE", "laptop1"), nodes, false, false, null); assertNotNull(nodeIds); assertEquals(1, nodeIds.size()); assertEquals("rgn1", nodeIds.iterator().next()); }
@Test public void testRouteS1ToDWFromRegsvr() { IDataRouter router = buildTestableRouter( MULTIPLE_GROUPS_PLUS_REG_SVR_NETWORKED_ROOT.findNetworkedNode("regsvr").getNode(), MULTIPLE_GROUPS_PLUS_REG_SVR_LINKS, MULTIPLE_GROUPS_PLUS_REG_SVR_NETWORKED_ROOT); Set<Node> nodes = new HashSet<Node>(); nodes.add(MULTIPLE_GROUPS_PLUS_REG_SVR_NETWORKED_ROOT.findNetworkedNode("s1").getNode()); nodes.add(MULTIPLE_GROUPS_PLUS_REG_SVR_NETWORKED_ROOT.findNetworkedNode("s2").getNode()); nodes.add(MULTIPLE_GROUPS_PLUS_REG_SVR_NETWORKED_ROOT.findNetworkedNode("dw").getNode()); Collection<String> nodeIds = router.routeToNodes(new SimpleRouterContext(), buildDataMetaData("SYM_NODE", "s1"), nodes, false, false, null); assertNotNull(nodeIds); assertEquals(2, nodeIds.size()); assertTrue(nodeIds.contains("s1")); assertTrue(nodeIds.contains("dw")); }
@Test public void testRouteDWToS1andS2FromRegsvr() { IDataRouter router = buildTestableRouter( MULTIPLE_GROUPS_PLUS_REG_SVR_NETWORKED_ROOT.findNetworkedNode("regsvr").getNode(), MULTIPLE_GROUPS_PLUS_REG_SVR_LINKS, MULTIPLE_GROUPS_PLUS_REG_SVR_NETWORKED_ROOT); Set<Node> nodes = new HashSet<Node>(); nodes.add(MULTIPLE_GROUPS_PLUS_REG_SVR_NETWORKED_ROOT.findNetworkedNode("s1").getNode()); nodes.add(MULTIPLE_GROUPS_PLUS_REG_SVR_NETWORKED_ROOT.findNetworkedNode("s2").getNode()); nodes.add(MULTIPLE_GROUPS_PLUS_REG_SVR_NETWORKED_ROOT.findNetworkedNode("dw").getNode()); Collection<String> nodeIds = router.routeToNodes(new SimpleRouterContext(), buildDataMetaData("SYM_NODE", "dw"), nodes, false, false, null); assertNotNull(nodeIds); assertEquals(3, nodeIds.size()); assertTrue(nodeIds.contains("s1")); assertTrue(nodeIds.contains("s2")); assertTrue(nodeIds.contains("dw")); }
@Test public void testRouteS1toRegsvrFromS1() { IDataRouter router = buildTestableRouter( MULTIPLE_GROUPS_PLUS_REG_SVR_NETWORKED_ROOT.findNetworkedNode("s1").getNode(), MULTIPLE_GROUPS_PLUS_REG_SVR_LINKS, MULTIPLE_GROUPS_PLUS_REG_SVR_NETWORKED_ROOT); Set<Node> nodes = new HashSet<Node>(); nodes.add(MULTIPLE_GROUPS_PLUS_REG_SVR_NETWORKED_ROOT.findNetworkedNode("s1").getNode()); nodes.add(MULTIPLE_GROUPS_PLUS_REG_SVR_NETWORKED_ROOT.findNetworkedNode("dw").getNode()); nodes.add(MULTIPLE_GROUPS_PLUS_REG_SVR_NETWORKED_ROOT.findNetworkedNode("regsvr").getNode()); Collection<String> nodeIds = router.routeToNodes(new SimpleRouterContext(), buildDataMetaData("SYM_NODE", "s1"), nodes, false, false, null); assertNotNull(nodeIds); assertEquals(1, nodeIds.size()); assertTrue(nodeIds.contains("regsvr")); }