@Test public void testStartReconnect() { final Long cookie = 0L; final Long shard = cookie; final ShardBackendInfo info = new ShardBackendInfo(clientActorContext().self(), 0L, ABIVersion.current(), SHARD_NAME, UnsignedLong.ZERO, Optional.of(tree), 10); final ConnectedClientConnection<ShardBackendInfo> newConn = AccessClientUtil.createConnectedConnection( clientActorContext(), cookie, info); object().createSnapshotProxy(TRANSACTION_ID, shard); final HistoryReconnectCohort reconnectCohort = object().startReconnect(newConn); Assert.assertNotNull(reconnectCohort); }
HistoryReconnectCohort startReconnect(final ConnectedClientConnection<ShardBackendInfo> newConn) { final ProxyHistory oldProxy; final long stamp = lock.writeLock(); try { oldProxy = histories.get(newConn.cookie()); } finally { lock.unlockWrite(stamp); } if (oldProxy == null) { return null; } final ProxyReconnectCohort proxy = verifyNotNull(oldProxy.startReconnect(newConn)); return new HistoryReconnectCohort() { @Override ProxyReconnectCohort getProxy() { return proxy; } @Override void replayRequests(final Collection<ConnectionEntry> previousEntries) { proxy.replayRequests(previousEntries); } @Override public void close() { LOG.debug("Client history {} finishing reconnect to {}", AbstractClientHistory.this, newConn); final ProxyHistory newProxy = proxy.finishReconnect(); if (!histories.replace(newConn.cookie(), oldProxy, newProxy)) { LOG.warn("Failed to replace proxy {} with {} in {}", oldProxy, newProxy, AbstractClientHistory.this); } } }; }
AbstractClientHistory extends LocalAbortable implements Identifiable<LocalHistoryIdentifier> { HistoryReconnectCohort startReconnect(final ConnectedClientConnection<ShardBackendInfo> newConn) { final ProxyHistory oldProxy; final long stamp = lock.writeLock(); try { oldProxy = histories.get(newConn.cookie()); } finally { lock.unlockWrite(stamp); } if (oldProxy == null) { return null; } final ProxyReconnectCohort proxy = verifyNotNull(oldProxy.startReconnect(newConn)); return new HistoryReconnectCohort() { @Override ProxyReconnectCohort getProxy() { return proxy; } @Override void replayRequests(final Collection<ConnectionEntry> previousEntries) { proxy.replayRequests(previousEntries); } @Override public void close() { LOG.debug("Client history {} finishing reconnect to {}", AbstractClientHistory.this, newConn); final ProxyHistory newProxy = proxy.finishReconnect(); if (!histories.replace(newConn.cookie(), oldProxy, newProxy)) { LOG.warn("Failed to replace proxy {} with {} in {}", oldProxy, newProxy, AbstractClientHistory.this); } } }; } }
AbstractClientHistory extends LocalAbortable implements Identifiable<LocalHistoryIdentifier> { HistoryReconnectCohort startReconnect(final ConnectedClientConnection<ShardBackendInfo> newConn) { final ProxyHistory oldProxy; final long stamp = lock.writeLock(); try { oldProxy = histories.get(newConn.cookie()); } finally { lock.unlockWrite(stamp); } if (oldProxy == null) { return null; } final ProxyReconnectCohort proxy = verifyNotNull(oldProxy.startReconnect(newConn)); return new HistoryReconnectCohort() { @Override ProxyReconnectCohort getProxy() { return proxy; } @Override void replayRequests(final Collection<ConnectionEntry> previousEntries) { proxy.replayRequests(previousEntries); } @Override public void close() { LOG.debug("Client history {} finishing reconnect to {}", AbstractClientHistory.this, newConn); final ProxyHistory newProxy = proxy.finishReconnect(); if (!histories.replace(newConn.cookie(), oldProxy, newProxy)) { LOG.warn("Failed to replace proxy {} with {} in {}", oldProxy, newProxy, AbstractClientHistory.this); } } }; } AbstractClientHistory(final AbstractDataStoreClientBehavior client, final LocalHistoryIdentifier identifier); }
AbstractClientHistory extends LocalAbortable implements Identifiable<LocalHistoryIdentifier> { HistoryReconnectCohort startReconnect(final ConnectedClientConnection<ShardBackendInfo> newConn) { final ProxyHistory oldProxy; final long stamp = lock.writeLock(); try { oldProxy = histories.get(newConn.cookie()); } finally { lock.unlockWrite(stamp); } if (oldProxy == null) { return null; } final ProxyReconnectCohort proxy = verifyNotNull(oldProxy.startReconnect(newConn)); return new HistoryReconnectCohort() { @Override ProxyReconnectCohort getProxy() { return proxy; } @Override void replayRequests(final Collection<ConnectionEntry> previousEntries) { proxy.replayRequests(previousEntries); } @Override public void close() { LOG.debug("Client history {} finishing reconnect to {}", AbstractClientHistory.this, newConn); final ProxyHistory newProxy = proxy.finishReconnect(); if (!histories.replace(newConn.cookie(), oldProxy, newProxy)) { LOG.warn("Failed to replace proxy {} with {} in {}", oldProxy, newProxy, AbstractClientHistory.this); } } }; } AbstractClientHistory(final AbstractDataStoreClientBehavior client, final LocalHistoryIdentifier identifier); @Override LocalHistoryIdentifier getIdentifier(); ClientTransaction createTransaction(); ClientSnapshot takeSnapshot(); }
AbstractClientHistory extends LocalAbortable implements Identifiable<LocalHistoryIdentifier> { HistoryReconnectCohort startReconnect(final ConnectedClientConnection<ShardBackendInfo> newConn) { final ProxyHistory oldProxy; final long stamp = lock.writeLock(); try { oldProxy = histories.get(newConn.cookie()); } finally { lock.unlockWrite(stamp); } if (oldProxy == null) { return null; } final ProxyReconnectCohort proxy = verifyNotNull(oldProxy.startReconnect(newConn)); return new HistoryReconnectCohort() { @Override ProxyReconnectCohort getProxy() { return proxy; } @Override void replayRequests(final Collection<ConnectionEntry> previousEntries) { proxy.replayRequests(previousEntries); } @Override public void close() { LOG.debug("Client history {} finishing reconnect to {}", AbstractClientHistory.this, newConn); final ProxyHistory newProxy = proxy.finishReconnect(); if (!histories.replace(newConn.cookie(), oldProxy, newProxy)) { LOG.warn("Failed to replace proxy {} with {} in {}", oldProxy, newProxy, AbstractClientHistory.this); } } }; } AbstractClientHistory(final AbstractDataStoreClientBehavior client, final LocalHistoryIdentifier identifier); @Override LocalHistoryIdentifier getIdentifier(); ClientTransaction createTransaction(); ClientSnapshot takeSnapshot();  }