RelayPullThread extends BasePullThread { public DatabusRelayConnection getLastOpenConnection() { return _lastOpenConnection; } RelayPullThread(String name,
                         DatabusSourcesConnection sourcesConn,
                         DbusEventBuffer dbusEventBuffer,
                         ConnectionStateFactory connStateFactory,
                         Set<ServerInfo> relays,
                         List<DbusKeyCompositeFilterConfig> relayFilterConfigs,
                         boolean isConsumeCurrent,
                         boolean isReadLatestScnOnErrorEnabled,
                         double pullerBufferUtilPct,
                         int noEventsConnectionResetTimeSec,
                         MBeanServer mbeanServer,
                         DbusEventFactory eventFactory,
                         Logger log); void setNoEventsConnectionResetTimeSec(long noEventsConnectionResetTimeSec); DatabusRelayConnection getLastOpenConnection();  }
@Test public void testRelayPullerThreadV3WithFilter() throws Exception { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); DbusKeyCompositeFilterConfig filterConfig = createFakeFilter(); bldr.setFilterConfig(filterConfig); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); MockRelayConnection conn = (MockRelayConnection) relayPuller.getLastOpenConnection(); conn.setProtocolVersion(3); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.REQUEST_SOURCES, StateId.SOURCES_RESPONSE_SUCCESS); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.SOURCES_RESPONSE_SUCCESS, StateId.REQUEST_REGISTER); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.REQUEST_REGISTER, StateId.REGISTER_RESPONSE_SUCCESS); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.REGISTER_RESPONSE_SUCCESS, StateId.REQUEST_STREAM); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.REQUEST_STREAM, StateId.STREAM_REQUEST_SUCCESS); DbusKeyCompositeFilter filter = conn.getRelayFilter(); Assert.assertNotNull(filter); KeyFilterConfigHolder configHolder = filterConfig.getConfigMap().values().iterator().next(); DbusKeyFilter inFilter = new DbusKeyFilter(configHolder); Assert.assertEquals(filter.getFilterMap().size(), 1, "1 filter in the composite filter"); Assert.assertTrue(filter.getFilterMap().entrySet().iterator().hasNext(), "has one filter"); DbusKeyFilter newFilter = filter.getFilterMap().values().iterator().next(); boolean eq = inFilter.getPartitionType() == newFilter.getPartitionType() && inFilter.getFilters().size() == newFilter.getFilters().size(); Assert.assertTrue(eq, "Same Filter"); }
@Test public void testShutdownRace() throws Exception { final Logger log = Logger.getLogger("TestRelayPullThread.testShutdownRace"); log.setLevel(Level.INFO); log.info("start"); List<String> sources = Arrays.asList("source1"); Properties clientProps = new Properties(); clientProps.setProperty("client.container.httpPort", "0"); clientProps.setProperty("client.runtime.bootstrap.enabled", "false"); clientProps.setProperty("client.runtime.relay(1).name", "relay1"); clientProps.setProperty("client.runtime.relay(1).port", "10001"); clientProps.setProperty("client.runtime.relay(1).sources", "source1"); clientProps.setProperty("client.connectionDefaults.eventBuffer.maxSize", "100000"); clientProps.setProperty("client.connectionDefaults.pullerRetries.maxRetryNum", "9"); clientProps.setProperty("client.connectionDefaults.pullerRetries.sleepIncFactor", "1.0"); clientProps.setProperty("client.connectionDefaults.pullerRetries.sleepIncDelta", "1"); clientProps.setProperty("client.connectionDefaults.pullerRetries.initSleep", "1"); DatabusHttpClientImpl client = new DatabusHttpClientImpl("client.", clientProps); Assert.assertNotNull(client, "client instantiation ok"); final DatabusHttpClientImpl.StaticConfig clientConf = client.getClientStaticConfig(); final DatabusSourcesConnection.StaticConfig srcConnConf = clientConf.getConnectionDefaults(); DatabusHttpClientImpl.RuntimeConfig clientRtConf = clientConf.getRuntime().build(); DbusEventBuffer.StaticConfig bufferConf = clientConf.getConnectionDefaults().getEventBuffer(); DbusEventBuffer relayBuffer = new DbusEventBuffer(bufferConf); DbusEventBuffer bootstrapBuffer = new DbusEventBuffer(bufferConf); AtomicInteger serverIdx = new AtomicInteger(-1); Set<ServerInfo> relays = clientRtConf.getRelaysSet(); List<ServerInfo> relayOrder = new ArrayList<ServerInfo>(relays); if (LOG.isInfoEnabled()) { StringBuilder sb = new StringBuilder(); for (ServerInfo serverInfo: relayOrder) { sb.append(serverInfo.getName()); sb.append(" "); } LOG.info("Relay order:" + sb.toString()); } List<IdNamePair> sourcesResponse = new ArrayList<IdNamePair>(); sourcesResponse.add(new IdNamePair(1L, "source1")); RegisterResponseEntry rre1 = new RegisterResponseEntry(1L, (short)1, SCHEMA$.toString()); final HashMap<Long, List<RegisterResponseEntry>> registerResponse = new HashMap<Long, List<RegisterResponseEntry>>(); registerResponse.put(1L, Arrays.asList(rre1)); final MockRelayConnection relayConn1 = new MockRelayConnection(sourcesResponse, registerResponse, null, serverIdx); final MockRelayConnection relayConn2 = new MockRelayConnectionForTestShutdownRace(sourcesResponse, null, null, serverIdx, log); DatabusRelayConnectionFactory mockConnFactory = EasyMock.createMock("mockRelayFactory", DatabusRelayConnectionFactory.class); EasyMock.expect(mockConnFactory.createRelayConnection( serverNameMatcher(serverIdx, relayOrder), EasyMock.<ActorMessageQueue>notNull(), EasyMock.<RemoteExceptionHandler>notNull())).andReturn(relayConn1); EasyMock.expect(mockConnFactory.createRelayConnection( serverNameMatcher(serverIdx, relayOrder), EasyMock.<ActorMessageQueue>notNull(), EasyMock.<RemoteExceptionHandler>notNull())).andReturn(relayConn2); EasyMock.replay(mockConnFactory); List<DatabusSubscription> sourcesSubList = DatabusSubscription.createSubscriptionList(sources); ConnectionStateFactory connStateFactory = new ConnectionStateFactory(sources); DatabusSourcesConnection sourcesConn = new DatabusSourcesConnection( srcConnConf, sourcesSubList, relays, null, null, null, relayBuffer, bootstrapBuffer, Executors.newCachedThreadPool(), null, null, null, null, null, null, null, mockConnFactory, null, null, null, null, new DbusEventV1Factory(), connStateFactory); final RelayPullThread relayPuller = new RelayPullThread("RelayPuller", sourcesConn, relayBuffer, connStateFactory, relays, new ArrayList<DbusKeyCompositeFilterConfig>(), !clientConf.getRuntime().getBootstrap().isEnabled(), clientConf.isReadLatestScnOnErrorEnabled(), clientConf.getPullerBufferUtilizationPct(), Integer.MAX_VALUE, ManagementFactory.getPlatformMBeanServer(), new DbusEventV1Factory(), null); RemoteExceptionHandler mockRemoteExceptionHandler = new MockRemoteExceptionHandler(sourcesConn, relayBuffer, relayPuller); Field field = relayPuller.getClass().getDeclaredField("_remoteExceptionHandler"); field.setAccessible(true); field.set(relayPuller, mockRemoteExceptionHandler); relayConn1.setCallback(relayPuller); relayConn2.setCallback(relayPuller); final Thread relayPullerThread = new Thread(relayPuller, "testShutdownRace.RelayPuller"); relayPullerThread.setDaemon(true); relayPullerThread.start(); relayPuller.enqueueMessage(LifecycleMessage.createStartMessage()); TestUtil.assertWithBackoff(new ConditionCheck() { @Override public boolean check() { return null != relayConn1.getLastStateMsg(); } }, "wait for call from the puller to the relay connection", 500, log); TestUtil.assertWithBackoff(new ConditionCheck() { @Override public boolean check() { log.debug(relayPuller.getMessageHistoryLog()); return !relayPullerThread.isAlive(); } }, "wait for puller to shutdown", 1000, log); EasyMock.verify(mockConnFactory); Assert.assertEquals(relayPuller.getLastOpenConnection(), null); log.info("done"); }