| 368 | } |
| 369 | |
| 370 | void connect(const id::UUID& _connectionId) |
| 371 | { |
| 372 | // It is possible that a new master was detected while we were waiting |
| 373 | // to establish a connection with the old master. |
| 374 | if (connectionId != _connectionId) { |
| 375 | VLOG(1) << "Ignoring connection attempt from stale connection"; |
| 376 | return; |
| 377 | } |
| 378 | |
| 379 | CHECK_EQ(DISCONNECTED, state); |
| 380 | CHECK_SOME(master); |
| 381 | |
| 382 | state = CONNECTING; |
| 383 | |
| 384 | auto connector = [this]() -> Future<Connection> { |
| 385 | return process::http::connect(master.get()); |
| 386 | }; |
| 387 | |
| 388 | // We create two persistent connections here, one for subscribe |
| 389 | // call/streaming response and another for non-subscribe calls/responses. |
| 390 | collect(connector(), connector()) |
| 391 | .onAny(defer(self(), &Self::connected, connectionId.get(), lambda::_1)); |
| 392 | } |
| 393 | |
| 394 | void connected( |
| 395 | const id::UUID& _connectionId, |