Subscription API unable to notify only some subscribers per schema subscription?

[Sarcasm]

Assuming a Subscription from a schema like this:

type Subscription {
    onEntityChanged(id: ID!) : Entity
}

I'd like a client to be notified only about the changes on the entity, or entities, it chose with id:

subscription EntityChanged {
     onEntityChanged(id: "someid") : Entity
}

I am under the impression that the current deliver mechanism does not allow for this granularity.

I can do things like service->deliver("onEntityChanged", nullptr); to notify when any entity changed, and getOnEntityChanged(params, id) has to generate a value for it. However, I'd like the subscription to notify only about the specified id, not about any id.

Did I miss something or is it a limitation of the current implementation?

[wravery]

You're right, that is a limitation. Right now it just visits the top level fields and figures out which ones are referenced in the subscription SelectionSet, but it leaves evaluating the arguments till delivery and query execution time.

The arguments are all known at the point when the subscription is created. It should be possible to visit them in SubscriptionDefinitionVisitor and build a response::Type::Map object of the arguments for comparison in SubscriptionData, much like fieldNames. Then deliver would need to take an additional (optional or default) parameter with arguments that must match (e.g. as another response::Type::Map object) to be included in any subscription callbacks.

Currently I think you could work around this by throwing a schema_exception from getOnEntityChanged(params, id) if the id doesn't match the object, and then in your callback you could ignore results that have a null data member and non-empty errors member. You could even set a specific error message for this case and look for a match in the callback function to make sure you're only ignoring this type of error.

[wravery]

Since you can have duplicate field names in the same query using field aliases, it also needs to keep a vector/list of arguments that match for each field name. The current implementation gets by with a std::unordered_set<> because it only cares if the field name was included, not whether or not any of the arguments matched.

[wravery]

Come to think of it, it also doesn't handle fragment spreads or inline fragments in SubscriptionDefinitionVisitor. Those would be on the subscription object itself and not on the field value SelectionSets, so it's less useful but it's theoretically allowed.

[wravery]

All set. 👍

[Sarcasm]

Great, thank you once again! :+1:

One remark. Do you think it should be possible to have the generated service provide a custom "filter" / "accept" method? id is one example of a filter that fully match. But I assume in theory the filter could be dependent on some client state or special matching method, or a combination of both.

For example for a filesystem watcher API, we could imagine something along these lines:

schema:

type Subscription {
    onFileAddedToDirectory(dir: String!, nameFilters: [String!]) : String!
}

subscription:

subscription CppFileAddedToDirectory {
     onFileAddedToDirectory(dir: "/tmp/blah", nameFilters=["*.h", "*.cpp"]) : String!
}

The subscriber, should be able to notify only changes that matches one of the glob patterns.

I do think the current API is simple and useful but it could be nice to provide extension points for customizable filtering behavior (e.g. some sort of bool accept() method that we could implement in the service. Or maybe the resolver could return some sort of sentinel value, something cleaner than throwing schema_exception. Could a default constructed std::future work as a sentinel?

In Request::deliver(): https://github.com/Microsoft/cppgraphqlgen/blob/d9b8f5fe3fbc00d1903355fe0058af306c6d74ea/GraphQLService.cpp#L1782

It may be possible to not call the callback if the resolve returns a "non-valid" future:

try {
    auto data = optionalOrDefaultSubscription->resolve(
        selectionSetParams, registration->selection,
        registration->data->fragments, registration->data->variables);

    if (!data.valid()) {
        continue;
    }

    result =
        std::async(std::launch::deferred,
                   [registration](std::future<response::Value> data) {
                       response::Value document(response::Type::Map);

                       document.emplace_back("data", data.get());

                       return document;
                   },
                   std::move(data));
} catch (...) {
    ...
}

[wravery]

I like your suggestion to check valid() on the future. That seems like a nice clean way to signal that individual subscription updates should be skipped from the resolver without needing to change any method signatures or introduce a new exception type.

I may open another issue to do some exception handling cleanup for the other operation types. The top-level future returned from there is created by graphqlservice with std::async, so it should never be invalid unless the caller mishandles it. But individual resolvers might have a bug or even get reused between subscription code and query code and throw future_error. Object::resolve should probably check for valid(), or Request::resolve could just catch all std::exception types including future_error, PEGTL parse errors, and maybe the logic_error exceptions from resposne::Value, and bundle all of those up into the errors value, much like schema_exception.

[wravery]

Not sure yet how to support the matches on fuzzy arguments like glob patterns. I might need to generate a virtual comparator with a concrete default implementation (like accept in your suggestion) specifically on the subscription object and move the exact comparison in there. Then an implementer could override it as you suggest to perform fuzzy matching.

[Sarcasm]

Throwing ideas without thinking about it too much.

It could be useful to have a callback called at the time of the subscription, with the input parameters, similar to the resolver function but that do not need to return a future. In this function a directory watch can be added and the subscription key can be stored. Maybe this function should be able to "reject" a subscription if it is invalid.

Additionally, maybe deliver could have a new overload taking directly a subscription key (listener), so that selection logic can be entirely custom if it does not fit in the "topic" model.

With these 2 ideas combined:

bool Query::aboutToSubscribeOnFileAddedToDirectory(service::SubscriptionFieldParams&& params,
                                                   response::StringType&& dirArg) const override
{
    std::cout << "subscription key: " << params.subscription << "\n";

  if (!std::filesystem::exists(dirArg)) {
    // cannot watch non-existing directories, reject/close this subscription
    return false;
  }

  m_fileWatches.push_back(m_fsWatcher.watch(dirArg,
                          [service=params.service, key=params.subscription](const FileChange &c) {
    // how to transmit 'c' to the resolver?
    service->deliver(key);
}));

return true;
}

Well, to be honest, this looks like an entirely different method than deliver.

Or maybe it should be possible to make custom names for deliver. For example, my initial request:

subscription EntityChanged {
     onEntityChanged(id: "someid") : Entity
}

The custom topic name could be:

onEntityChanged.someid

However I'm not sure we can find suitable topic names for all sort of input arguments.

[wravery]

Fuzzy matching with #31. Rather than making it the responsibility of the subscription object, I made it a callback in the deliver method.

[Sarcasm]

Great, thank you again! :+1: Additionally, I think I understand the subscription object better now, which will be useful too.