CouchbaseConnectionFactory cf = connectionFactory;
CouchbaseConnection connection = null;
List<ConnectionObserver> initialObservers = new ArrayList<ConnectionObserver>();
final CountDownLatch latch = new CountDownLatch(1);
initialObservers.add(new ConnectionObserver() {
@Override
public void connectionEstablished(SocketAddress socketAddress, int i) {
latch.countDown();
}
@Override
public void connectionLost(SocketAddress socketAddress) {
// not needed
}
});
try {
connection = new CouchbaseConfigConnection(
cf.getReadBufSize(), fact, Collections.singletonList(node),
initialObservers, cf.getFailureMode(),
cf.getOperationFactory()
);
boolean result = latch.await(5, TimeUnit.SECONDS);
if (!result) {
throw new IOException("Connection could not be established to carrier"
+ " port in the given time interval.");
}
} catch (Exception ex) {
if (connection != null) {
connection.shutdown();
}
getLogger().debug("(Carrier Publication) Could not load config from "
+ node.getHostName() + ", trying next node.", ex);
return false;
}
if (!bucket.equals(ANONYMOUS_BUCKET)) {
AuthThreadMonitor monitor = new AuthThreadMonitor();
List<MemcachedNode> connectedNodes = new ArrayList<MemcachedNode>(
connection.getLocator().getAll());
for (MemcachedNode connectedNode : connectedNodes) {
if (connectedNode.getSocketAddress().equals(node)) {
monitor.authConnection(connection, cf.getOperationFactory(),
cf.getAuthDescriptor(), connectedNode);
}
}
}
List<String> configs = getConfigsFromBinaryConnection(connection);
if (configs.isEmpty()) {
getLogger().debug("(Carrier Publication) Could not load config from "
+ node.getHostName() + ", trying next node.");
connection.shutdown();
return false;
}
String appliedConfig = connection.replaceConfigWildcards(
configs.get(0));
try {
Bucket config = configurationParser.parseBucket(appliedConfig);
setConfig(config);
} catch(Exception ex) {
getLogger().warn("Could not parse config, retrying bootstrap.", ex);
connection.shutdown();
return false;
}
connection.addObserver(new ConnectionObserver() {
@Override
public void connectionEstablished(SocketAddress sa, int reconnectCount) {
getLogger().debug("Carrier Config Connection established to " + sa);
}