PulseEngine currently sends messages of the form {[eventName]: payload}. Pulse Inspector expects to also show the exchange, primary routing key, and additional routing keys (cc's). PulseEngine currently discards that information. Pulse Inspector also allows listening to arbitrary exchanges and routing-key patterns. I think the right thing to do is modify PulseEngine to call onMessage with all of the necessary information, and then filter out only the relevant information in the purpose-specific GraphQL resolvers. Is that about right? What I don't know is, what should the GraphQL subscriptions look like for Pulse Inspector?

> I think the right thing to do is modify PulseEngine to call onMessage with all of the necessary information, and then filter out only the relevant information in the purpose-specific GraphQL resolvers. Is that about right? That sounds about right. The server will only run the resolvers that are asked by the client. > What I don't know is, what should the GraphQL subscriptions look like for Pulse Inspector? The return types are defined in [1]. Also, basic subscription queries can be found in [2]. Let me know if that doesn't help answer your question. [1] https://github.com/taskcluster/taskcluster-web-server/blob/master/src/graphql/Tasks.graphql#L399-L422 [2] https://github.com/taskcluster/taskcluster-web-server#sample-subscriptions

From discussion, we'll have two "tiers" of subscriptions: - type-specific subscriptions (currently just for task events) - generic subscriptions (for pulse inspector)

So right now, subscriptions look like this: Subscription: { tasksDefined: { subscribe(parent, { taskGroupId }, { pulseEngine, clients }) { const routingKey = { taskGroupId }; const binding = clients.queueEvents.taskDefined(routingKey); return pulseEngine.asyncIterator('tasksDefined', [ { exchange: binding.exchange, pattern: binding.routingKeyPattern, }, ]); }, }, there's nowhere to plug in message-transformation function. What would be the stylistically ideal way to add that? AsyncIterator sounds like some fairly well-known abstraction, perhaps with a specific implementation in src/PulseEngine/AsyncIterator.js. If that's the case, is there a spec for it? I can find some proposals by googling, but I'm not sure which (if any) apply. Is there a `map` method we could apply? So return pulseEngine.asyncIterator('tasksDefined', [ { exchange: binding.exchange, pattern: binding.routingKeyPattern, }, ]).map(async message => ({ .. })); // <-- do the transform Otherwise, I could potentially add a transform function as an argument to pulseEngine.asyncIterator. What would you recommend?

`src/PulseEngine/AsyncIterator.js` is a wrapper around the AsyncIterator that implements custom logic on the subscriptions like adding static fields. There's an example in [1] that shows you how to manipulate the payload and some more information like the TC39 proposal. You can also manipulate the payload by adding a resolve method in the subscription resolver[2]. Hope that helps. [1] https://github.com/apollographql/graphql-subscriptions#custom-asynciterator-wrappers [2] https://github.com/apollographql/graphql-subscriptions#payload-manipulation