RelayPullThread extends BasePullThread { protected BackoffTimer getRetryonFallOff() { return _retriesOnFallOff; } 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(groups={"small", "functional"}) public void testRelayTransition() throws Exception { final Logger log = Logger.getLogger("TestRelayPullThread.testRelayTransition"); log.info("---------- start ---------------"); { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); RelayPullThread relayPuller = bldr.createRelayPullThread(); Assert.assertEquals(relayPuller.getConnectionState().getStateId(), StateId.INITIAL, "Initial State Check"); Thread relayPullerThread = new Thread(relayPuller); relayPullerThread.setDaemon(false); relayPullerThread.start(); relayPuller.enqueueMessage(LifecycleMessage.createStartMessage()); try { Thread.sleep(500); } catch (InterruptedException ie){} for (int i = 0 ; i < 100; i++ ) { try { Thread.sleep(5); Assert.assertTrue(relayPuller.getConnectionState().getStateId() != StateId.INITIAL,"StateId can never be INITIAL once started"); } catch (InterruptedException ie){} } relayPuller.enqueueMessage(LifecycleMessage.createShutdownMessage()); relayPuller.awaitShutdown(); } { { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(true, false); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.PICK_SERVER, "SUSPEND_ON_ERROR"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); relayPuller.getServers().clear(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.PICK_SERVER, "SUSPEND_ON_ERROR"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, true); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.REQUEST_SOURCES, StateId.SOURCES_REQUEST_SENT, ""); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.REQUEST_SOURCES, StateId.SOURCES_RESPONSE_SUCCESS); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, true); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.REQUEST_SOURCES, StateId.SOURCES_REQUEST_SENT, ""); List<IdNamePair> sourcesResponse = new ArrayList<IdNamePair>(); sourcesResponse.add(new IdNamePair(1L, "source1")); connState.switchToSourcesSuccess(sourcesResponse, _HOSTNAME, _SVCNAME); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.SOURCES_RESPONSE_SUCCESS, StateId.REQUEST_REGISTER); validateConnState(connState); String expSubsListStr = "[{\"physicalSource\":{\"uri\":\"databus:physical-source:ANY\",\"role\":\"ANY\"},\"physicalPartition\":{\"id\":-1,\"name\":\"*\"},\"logicalPartition\":{\"source\":{\"id\":1,\"name\":\"source1\"},\"id\":-1}}]"; String expSourcesIdListStr = "1"; Assert.assertEquals(connState.getSourcesIdListString(), expSourcesIdListStr, "SourcesId Added"); String subsListStr = connState.getSubsListString(); Assert.assertEquals(subsListStr, expSubsListStr); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, true); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.REQUEST_SOURCES, StateId.SOURCES_REQUEST_SENT, ""); List<IdNamePair> sourcesResponse = new ArrayList<IdNamePair>(); sourcesResponse.add(new IdNamePair(1L, "source1")); connState.switchToSourcesSuccess(sourcesResponse, _HOSTNAME, _SVCNAME); relayPuller.getMessageQueue().clear(); connState.getSourcesNameMap().clear(); testTransitionCase(relayPuller, StateId.SOURCES_RESPONSE_SUCCESS, StateId.PICK_SERVER); validateConnState(connState); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, true); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.REQUEST_SOURCES, StateId.SOURCES_REQUEST_SENT, ""); List<IdNamePair> sourcesResponse = new ArrayList<IdNamePair>(); sourcesResponse.add(new IdNamePair(1L, "source1")); connState.switchToSourcesSuccess(sourcesResponse, _HOSTNAME, _SVCNAME); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.SOURCES_RESPONSE_SUCCESS, StateId.REQUEST_REGISTER); relayPuller.getMessageQueue().clear(); String subsListString = "[{\"physicalSource\":{\"uri\":\"databus:physical-source:ANY\",\"role\":\"MASTER\"},\"physicalPartition\":{\"id\":-1,\"name\":\"*\"},\"logicalPartition\":{\"source\":{\"id\":0,\"name\":\"source1\"},\"id\":-1}}]"; String sourcesIdListString = "1"; connState.switchToRequestSourcesSchemas(sourcesIdListString, subsListString); testTransitionCase(relayPuller, StateId.REQUEST_REGISTER, StateId.REGISTER_REQUEST_SENT,""); validateConnState(connState); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); connState.getSourcesSchemas().clear(); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.REGISTER_RESPONSE_SUCCESS, StateId.PICK_SERVER); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); Assert.assertEquals(relayPuller.getConnectionState().isSCNRegress(), false, "SCN Regress check"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); RelayPullThread relayPuller = bldr.createRelayPullThread(); Checkpoint cp = new Checkpoint(); cp.setConsumptionMode(DbusClientMode.ONLINE_CONSUMPTION); cp.setWindowScn(100L); cp.setWindowOffset(20); cp.setPrevScn(80L); relayPuller.getConnectionState().setCheckpoint(cp); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); Assert.assertEquals(relayPuller.getConnectionState().getCheckpoint().getWindowScn(), 80L, "WindowSCN check"); Assert.assertEquals(relayPuller.getConnectionState().getCheckpoint().getWindowOffset(), new Long(-1), "WindowOffset check"); Assert.assertEquals(relayPuller.getConnectionState().isSCNRegress(), true, "SCN Regress check"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); RelayPullThread relayPuller = bldr.createRelayPullThread(); Checkpoint cp = new Checkpoint(); cp.setConsumptionMode(DbusClientMode.ONLINE_CONSUMPTION); cp.setWindowScn(100L); cp.setWindowOffset(20); relayPuller.getConnectionState().setCheckpoint(cp); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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.REGISTER_RESPONSE_SUCCESS, "SUSPEND_ON_ERROR"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); bldr.setBootstrapEnabled(true); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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(); Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), false, "Is Relay FellOff"); Checkpoint cp = new Checkpoint(); cp.setConsumptionMode(DbusClientMode.BOOTSTRAP_SNAPSHOT); connState.setCheckpoint(cp); testTransitionCase(relayPuller, StateId.REGISTER_RESPONSE_SUCCESS, StateId.BOOTSTRAP); Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), true, "Is Relay FellOff"); Assert.assertEquals(relayPuller.getConnectionState().isSCNRegress(), false, "SCN Regress check"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); bldr.setBootstrapEnabled(true); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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(); Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), false, "Is Relay FellOff"); Checkpoint cp = new Checkpoint(); cp.setConsumptionMode(DbusClientMode.BOOTSTRAP_CATCHUP); connState.setCheckpoint(cp); testTransitionCase(relayPuller, StateId.REGISTER_RESPONSE_SUCCESS, StateId.BOOTSTRAP); Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), true, "Is Relay FellOff"); Assert.assertEquals(relayPuller.getConnectionState().isSCNRegress(), false, "SCN Regress check"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, true); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.REQUEST_SOURCES, StateId.SOURCES_REQUEST_SENT, ""); List<IdNamePair> sourcesResponse = new ArrayList<IdNamePair>(); sourcesResponse.add(new IdNamePair(1L, "source1")); connState.switchToSourcesSuccess(sourcesResponse, _HOSTNAME, _SVCNAME); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.SOURCES_RESPONSE_SUCCESS, StateId.REQUEST_REGISTER); relayPuller.getMessageQueue().clear(); String subsListString = "[{\"physicalSource\":{\"uri\":\"databus:physical-source:ANY\",\"role\":\"MASTER\"},\"physicalPartition\":{\"id\":-1,\"name\":\"*\"},\"logicalPartition\":{\"source\":{\"id\":0,\"name\":\"source1\"},\"id\":-1}}]"; String sourcesIdListString = "1"; connState.switchToRequestSourcesSchemas(sourcesIdListString, subsListString); testTransitionCase(relayPuller, StateId.REQUEST_REGISTER, StateId.REGISTER_REQUEST_SENT,""); relayPuller.getMessageQueue().clear(); MockRelayConnection conn = (MockRelayConnection)connState.getRelayConnection(); connState.switchToRegisterSuccess(conn.getRegisterResponse(), conn.getRegisterResponse(), conn.getRegisterMetadataResponse()); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.REGISTER_RESPONSE_SUCCESS, StateId.REQUEST_STREAM); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.REQUEST_STREAM, StateId.STREAM_REQUEST_SENT, ""); validateConnState(connState); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.STREAM_RESPONSE_DONE); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); bldr.setBootstrapEnabled(true).setReadLatestScnEnabled(false).setReadDataThrowException(false) .setReadDataException(true).setExceptionName(ScnNotFoundException.class.getName()).setNumRetriesOnFellOff(5); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); for ( int i = 1; i <= 6 ; i++) { System.out.println("Iteration :" + i); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); relayPuller.getMessageQueue().clear(); if ( i < 6 ) { testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.PICK_SERVER); Assert.assertEquals(relayPuller.getRetryonFallOff().getRemainingRetriesNum(), 6-i ,"Retry State"); } else { testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.BOOTSTRAP); Assert.assertEquals(relayPuller.getRetryonFallOff().getRemainingRetriesNum(), 5 ,"Retry State"); } relayPuller.getMessageQueue().clear(); } Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), true, "RelayFellOff State"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); bldr.setBootstrapEnabled(true).setReadLatestScnEnabled(false).setReadDataThrowException(false) .setReadDataException(true).setExceptionName(ScnNotFoundException.class.getName()); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.BOOTSTRAP); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); bldr.setBootstrapEnabled(false).setReadLatestScnEnabled(true).setReadDataThrowException(false) .setReadDataException(true).setExceptionName(ScnNotFoundException.class.getName()).setNumRetriesOnFellOff(5); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); for (int i=1; i<=6;i++) { testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); relayPuller.getMessageQueue().clear(); if ( i < 6 ) { testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.PICK_SERVER); Assert.assertEquals(relayPuller.getRetryonFallOff().getRemainingRetriesNum(), 6-i ,"Retry State"); } else { testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.STREAM_RESPONSE_DONE); Assert.assertEquals(relayPuller.getRetryonFallOff().getRemainingRetriesNum(), 5 ,"Retry State"); } relayPuller.getMessageQueue().clear(); } MockRelayConnection conn = (MockRelayConnection)connState.getRelayConnection(); Assert.assertEquals(conn.isReadFromLatestScn(),true, "ReadFromLatestScn set"); Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), false, "RelayFellOff State"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); bldr.setBootstrapEnabled(false).setReadLatestScnEnabled(true).setReadDataThrowException(false) .setReadDataException(true).setExceptionName(ScnNotFoundException.class.getName()); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.STREAM_RESPONSE_DONE); MockRelayConnection conn = (MockRelayConnection)connState.getRelayConnection(); Assert.assertEquals(conn.isReadFromLatestScn(),true, "ReadFromLatestScn set"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); bldr.setBootstrapEnabled(false).setReadLatestScnEnabled(false).setReadDataThrowException(false) .setReadDataException(true).setExceptionName(ScnNotFoundException.class.getName()).setNumRetriesOnFellOff(5); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); for (int i = 1; i <= 6; i++) { testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); relayPuller.getMessageQueue().clear(); if ( i < 6 ) { testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.PICK_SERVER); Assert.assertEquals(relayPuller.getRetryonFallOff().getRemainingRetriesNum(), 6-i ,"Retry State"); } else { testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.STREAM_REQUEST_SUCCESS, "SUSPEND_ON_ERROR"); Assert.assertEquals(relayPuller.getRetryonFallOff().getRemainingRetriesNum(), 5 ,"Retry State"); } relayPuller.getMessageQueue().clear(); } Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), true, "RelayFellOff State"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); bldr.setBootstrapEnabled(false).setReadLatestScnEnabled(false).setReadDataThrowException(false) .setReadDataException(true).setExceptionName(ScnNotFoundException.class.getName()); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.STREAM_REQUEST_SUCCESS, "SUSPEND_ON_ERROR"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); bldr.setBootstrapEnabled(true).setReadLatestScnEnabled(false).setReadDataThrowException(false) .setReadDataException(true).setExceptionName("DummyError"); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.STREAM_RESPONSE_ERROR); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.STREAM_RESPONSE_DONE); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.STREAM_RESPONSE_DONE,StateId.REQUEST_STREAM); Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), false, "Is Relay FellOff"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); bldr.setBootstrapEnabled(true).setReadLatestScnEnabled(false).setReadDataThrowException(false) .setReadDataException(true).setExceptionName(ScnNotFoundException.class.getName()); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.BOOTSTRAP); relayPuller.getMessageQueue().clear(); Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), true, "Is Relay FellOff"); testTransitionCase(relayPuller, StateId.BOOTSTRAP,StateId.BOOTSTRAP_REQUESTED,""); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); bldr.setBootstrapEnabled(true).setReadLatestScnEnabled(false).setReadDataThrowException(true) .setReadDataException(true).setExceptionName(ScnNotFoundException.class.getName()); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); relayPuller.getMessageQueue().clear(); Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), false, "Is Relay FellOff"); testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.BOOTSTRAP); relayPuller.getMessageQueue().clear(); Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), true, "Is Relay FellOff"); testTransitionCase(relayPuller, StateId.BOOTSTRAP,StateId.BOOTSTRAP, "SUSPEND_ON_ERROR"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); bldr.setBootstrapEnabled(true).setReadLatestScnEnabled(false).setReadDataThrowException(false) .setReadDataException(true).setExceptionName(ScnNotFoundException.class.getName()); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.BOOTSTRAP); relayPuller.getMessageQueue().clear(); Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), true, "Is Relay FellOff"); testTransitionCase(relayPuller, StateId.BOOTSTRAP,StateId.BOOTSTRAP_REQUESTED,""); BootstrapResultMessage msg = BootstrapResultMessage.createBootstrapFailedMessage(new Exception("Dummy")); doExecuteAndChangeState(relayPuller,msg); Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), true, "Is Relay FellOff"); Assert.assertEquals(connState.getStateId(),StateId.BOOTSTRAP, "BOOTSTRAP_REQUESTED to BOOTSTRAP"); Assert.assertEquals(relayPuller.getQueueListString(), "RelayPuller queue: [BOOTSTRAP]", "Queue :BOOTSTRAP_REQUESTED to BOOTSTRAP"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); bldr.setBootstrapEnabled(true).setReadLatestScnEnabled(false).setReadDataThrowException(false) .setReadDataException(true).setExceptionName(ScnNotFoundException.class.getName()); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.BOOTSTRAP); relayPuller.getMessageQueue().clear(); Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), true, "Is Relay FellOff"); testTransitionCase(relayPuller, StateId.BOOTSTRAP,StateId.BOOTSTRAP_REQUESTED,""); BootstrapResultMessage msg = BootstrapResultMessage.createBootstrapCompleteMessage(null); doExecuteAndChangeState(relayPuller,msg); Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), true, "Is Relay FellOff"); Assert.assertEquals(connState.getStateId(),StateId.BOOTSTRAP, "BOOTSTRAP_REQUESTED to BOOTSTRAP"); Assert.assertEquals(relayPuller.getQueueListString(), "RelayPuller queue: [BOOTSTRAP]", "Queue :BOOTSTRAP_REQUESTED to BOOTSTRAP"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); bldr.setBootstrapEnabled(true).setReadLatestScnEnabled(false).setReadDataThrowException(false) .setReadDataException(true).setExceptionName(ScnNotFoundException.class.getName()); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); 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); relayPuller.getMessageQueue().clear(); testTransitionCase(relayPuller, StateId.STREAM_REQUEST_SUCCESS,StateId.BOOTSTRAP); relayPuller.getMessageQueue().clear(); Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), true, "Is Relay FellOff"); testTransitionCase(relayPuller, StateId.BOOTSTRAP,StateId.BOOTSTRAP_REQUESTED,""); BootstrapResultMessage msg = BootstrapResultMessage.createBootstrapCompleteMessage(new Checkpoint()); doExecuteAndChangeState(relayPuller,msg); Assert.assertEquals(relayPuller.getConnectionState().isRelayFellOff(), false, "Is Relay FellOff"); Assert.assertEquals(connState.getStateId(),StateId.REQUEST_STREAM, "BOOTSTRAP_REQUESTED to REQUEST_STREAM"); Assert.assertEquals(relayPuller.getQueueListString(), "RelayPuller queue: [REQUEST_STREAM]", "Queue :BOOTSTRAP_REQUESTED to REQUEST_STREAM"); } { RelayPullThreadBuilder bldr = new RelayPullThreadBuilder(false, false); bldr.setBootstrapEnabled(true).setReadLatestScnEnabled(false).setReadDataThrowException(false) .setReadDataException(true).setExceptionName(ScnNotFoundException.class.getName()); RelayPullThread relayPuller = bldr.createRelayPullThread(); relayPuller.getComponentStatus().start(); ConnectionState connState = relayPuller.getConnectionState(); connState.switchToPickServer(); testTransitionCase(relayPuller, StateId.PICK_SERVER, StateId.REQUEST_SOURCES); relayPuller.getMessageQueue().clear(); connState.switchToSourcesRequestError(); testTransitionCase(relayPuller, StateId.SOURCES_REQUEST_ERROR, StateId.PICK_SERVER); relayPuller.getMessageQueue().clear(); connState.switchToSourcesResponseError(); testTransitionCase(relayPuller, StateId.SOURCES_RESPONSE_ERROR, StateId.PICK_SERVER); relayPuller.getMessageQueue().clear(); connState.switchToRegisterRequestError(); testTransitionCase(relayPuller, StateId.REGISTER_REQUEST_ERROR, StateId.PICK_SERVER); relayPuller.getMessageQueue().clear(); connState.switchToRegisterResponseError(); testTransitionCase(relayPuller, StateId.REGISTER_RESPONSE_ERROR, StateId.PICK_SERVER); relayPuller.getMessageQueue().clear(); connState.switchToStreamRequestError(); testTransitionCase(relayPuller, StateId.STREAM_REQUEST_ERROR, StateId.PICK_SERVER); relayPuller.getMessageQueue().clear(); connState.switchToStreamResponseError(); testTransitionCase(relayPuller, StateId.STREAM_RESPONSE_ERROR, StateId.PICK_SERVER); } } log.info("--------- end -------------"); }