Method __send

src/scheduler/scheduler.cpp:625–720

  }

  void __send(
      const id::UUID& _connectionId,
      const Call& call,
      const Future<process::http::Response>& response)
  {
    // It is possible that we detected a new master before a response could
    // be received.
    if (connectionId != _connectionId) {
      return;
    }

    CHECK(!response.isDiscarded());
    CHECK(state == SUBSCRIBING || state == SUBSCRIBED) << state;

    // This can happen during a master failover or a network blip
    // causing the socket to timeout. Eventually, the scheduler would
    // detect the disconnection via ZK(disconnect()) or lack of heartbeats.
    if (response.isFailed()) {
      LOG(ERROR) << "Request for call type " << call.type() << " failed: "
                 << response.failure();
      return;
    }

    if (response->code == process::http::Status::OK) {
      // Only SUBSCRIBE call should get a "200 OK" response.
      CHECK_EQ(Call::SUBSCRIBE, call.type());
      CHECK_EQ(response->type, process::http::Response::PIPE);
      CHECK_SOME(response->reader);

      state = SUBSCRIBED;

      Pipe::Reader reader = response->reader.get();

      Owned<Reader<Event>> decoder(new Reader<Event>(
          lambda::bind(deserialize<Event>, contentType, lambda::_1),
          reader));

      subscribed = SubscribedResponse {reader, decoder};

      // Responses to SUBSCRIBE calls should always include a stream ID.
      CHECK(response->headers.contains("Mesos-Stream-Id"));

      Try<id::UUID> uuid =
        id::UUID::fromString(response->headers.at("Mesos-Stream-Id"));

      CHECK_SOME(uuid);

      streamId = uuid.get();

      read();

      return;
    }

    if (response->code == process::http::Status::ACCEPTED) {
      // Only non SUBSCRIBE calls should get a "202 Accepted" response.
      CHECK_NE(Call::SUBSCRIBE, call.type());
      return;