PsqlEventSource implements EventSource { @Override public void connect() throws DataAccessException { if (!isConnected()) { connection = connectionProvider.getConnection(); } subscribeToStream(); } PsqlEventSource(String dbUrl, String username, String password, String tableName);  PsqlEventSource(String dbUrl, String username, String password, String tableName, boolean autoConfig);  PsqlEventSource(ConnectionProvider connectionProvider, String tableName);  PsqlEventSource(ConnectionProvider connectionProvider, String tableName, boolean autoConfig); @VisibleForTesting  PsqlEventSource(ConnectionProvider connectionProvider, String tableName, PsqlEventMapper mapper,
                                       boolean autoConfig); @VisibleForTesting  PsqlEventSource(ConnectionProvider connectionProvider, String tableName, PsqlEventMapper mapper,
                           boolean autoConfig, PsqlConfigurator configurator); @Override List<Event<Map<String, Object>>> getNewEvents(); @Override void connect(); @Override void disconnect(); @Override boolean isConnected(); @Override void setup(); @Override void cleanup();  }
@Test(groups = SMALL) public void testConnectUsesConnectionProviderToGetDbConnection() { eventSource.connect(); verify(connectionProvider, times(2)).getConnection(); }
@Test(groups = SMALL) public void testConnectDoesntGetANewConnectionIfOneAlreadyExistsAndIsOpen() { eventSource.connect(); eventSource.connect(); verify(connectionProvider, times(2)).getConnection(); }
@Test(groups = SMALL) public void testConnectCreatesANewConnectionTheExistingWasClosed() throws SQLException { when(mockedConnection.isClosed()).thenReturn(true); eventSource.connect(); eventSource.connect(); verify(connectionProvider, times(3)).getConnection(); }
@Test(groups = SMALL, expectedExceptions = DataAccessException.class) public void testConnectThrowsDataAccessExceptionWhenCanNotGetConnectionToTheDb() { when(connectionProvider.getConnection()).thenThrow(new DataAccessException("")); eventSource.connect(); }
@Test(groups = SMALL) public void testConnectRegistersAListenerInPsqlForGivenStreamName() throws SQLException { Statement mockedStmt = mock(Statement.class); when(mockedConnection.createStatement()).thenReturn(mockedStmt); eventSource.connect(); verify(mockedStmt).execute("LISTEN " + STREAM_NAME); }