Method _subscribe

src/resource_provider/manager.cpp:847–930

void ResourceProviderManagerProcess::_subscribe(
    const Future<bool>& admitResourceProvider,
    Owned<ResourceProvider> resourceProvider)
{
  if (!admitResourceProvider.isReady()) {
    LOG(INFO)
      << "Not subscribing resource provider " << resourceProvider->info.id()
      << " as registry update did not succeed: " << admitResourceProvider;

    return;
  }

  CHECK(admitResourceProvider.get())
    << "Could not admit resource provider " << resourceProvider->info.id()
    << " as registry update was rejected";

  const ResourceProviderID& resourceProviderId = resourceProvider->info.id();

  resourceProvider->http.closed()
    .onAny(defer(self(), [=](const Future<Nothing>& future) {
      // Iff the remote side closes the HTTP connection, the future will be
      // ready. We will remove the resource provider in that case.
      // This side closes the HTTP connection only when removing a resource
      // provider, therefore we shouldn't try to remove it again here.
      if (future.isReady()) {
        CHECK(resourceProviders.subscribed.contains(resourceProviderId));

        // NOTE: All pending futures of publish requests for the resource
        // provider will become failed.
        resourceProviders.subscribed.erase(resourceProviderId);
      }

      ResourceProviderMessage::Disconnect disconnect{resourceProviderId};

      ResourceProviderMessage message;
      message.type = ResourceProviderMessage::Type::DISCONNECT;
      message.disconnect = std::move(disconnect);

      messages.put(std::move(message));

      ++metrics.disconnections;
    }));

  if (!resourceProviders.known.contains(resourceProviderId)) {
    mesos::resource_provider::registry::ResourceProvider resourceProvider_ =
      createRegistryResourceProvider(resourceProvider->info);

    resourceProviders.known.put(
        resourceProviderId,
        std::move(resourceProvider_));
  }

  ResourceProviderMessage::Subscribe subscribe{resourceProvider->info};

  ResourceProviderMessage message;
  message.type = ResourceProviderMessage::Type::SUBSCRIBE;
  message.subscribe = std::move(subscribe);