Help getting subscriptions working

[robinelvin]

Hi,

I am trying to add subscriptions to my schema and test them before enabling in production. I am using graphql-redis-subscriptions and I am passing in a shared pubSub object into my buildSchema call. I can subscribe using the pubSub.subscribe method and when I use pubSub.publish I get the message. Next I am trying to subscribe using a GraphQL subscription query. The subscription fires the topics method on the @Subscription decorator and I can confirm the topic is being passed correctly. The problem is I fire the pubSub.publish as before and I get nothing. The test just times out and the resolver method is never called. I have used both the subscribe method from the graphql package and tried wiring up a websocket server. What am I missing?

/** Export schema generated with TypeGraphQL */
import 'reflect-metadata';
import { buildSchema } from 'type-graphql';
import { getPubSub } from '../../../domain/services/RedisPubSub';

import {
    OrganisationResolver,
    UserResolver,
    LoginResolver,
    UserManagementResolver,
    ProcessResolver,
    ModuleResolver,
    SyncEventResolver,
    EventResolver,
} from './resolvers/resolvers';

export const pubSub = getPubSub();

export default buildSchema({
    resolvers: [
        OrganisationResolver,
        ProcessResolver,
        UserResolver,
        LoginResolver,
        UserManagementResolver,
        ModuleResolver,
        SyncEventResolver,
        EventResolver,
    ],
    dateScalarMode: 'isoDate',
    pubSub,
});

/** Part of SyncEventResolver **/
@Subscription({
        topics: ({ args, payload, context }) => args.topic,
    })
    SyncEvents(
        @Arg('topic') topic: string,
        @Root() payload: SyncEvent
    ): SyncEvent {
        // *** THIS IS NEVER CALLED ***
        console.log('SUBSCRIPTION', payload, topic);
        let msg: SyncEvent;
        switch (payload.type) {
            case 'RolesAddedToUser':
                msg = new RolesAddedToUser();
                break;
            case 'RolesRemovedFromUser':
                msg = new RolesRemovedFromUser();
                break;
            case 'UserCreated':
                msg = new UserCreated();
                break;
            case 'UserNameChanged':
                msg = new UserNameChanged();
                break;
            default:
                msg = payload;
        }

        // Copy properties
        Object.keys(payload).forEach((prop) => {
            msg[prop] = payload[prop];
        });
        return msg;
    }

/** Test **/
import {
    IDesignsetArchivedChangedSyncEvent,
    ISyncEvent,
} from './common';
import {
    execute,
    graphql,
    GraphQLSchema,
    subscribe,
    parse,
    ExecutionResult,
} from 'graphql';
import buildSchema, {
    pubSub,
} from '../../../../source/application/data-access/graphql/typedschema';
import { mockContext } from '../repositories/MockContext';

describe('Subscriptions', () => {
    let schema: GraphQLSchema;

    beforeAll(async () => {
        schema = await buildSchema;
    });

    afterAll(() => {
        pubSub.close();
    });

    describe('Sync Events', () => {
        const subscription = `
            subscription SyncEvents($topic: String!) {
        SyncEvents(topic: $topic) {
            userId
            eventId
            timestamp
            type

            ... on UserNameChanged {
                id
                oldFirstName
                oldLastName
                newFirstName
                newLastName
            }

            ... on UserCreated {
                id
                firstName
                lastName
                emailAddress
                isSuperuser
                organisationMembership {
                    identifier
                    organisation {
                        id
                        name
                        logoUrl
                    }
                }
            }

            ... on RolesAddedToUser {
                id
                rolesAdded {
                    ...roleFields
                }
            }

            ... on RolesRemovedFromUser {
                id
                rolesRemoved {
                    ...roleFields
                }
            }

        }
    }
        `;

        it('should emit UserNameChanged', async () => {
            const event: IUserNameChangedSyncEvent = {
                eventId: '1',
                id: 'userchanged-1',
                oldFirstName: 'Wrong',
                newFirstName: 'Right',
                oldLastName: 'Bad',
                newLastName: 'Good',
                timestamp: '2022-11-28T10:19:00',
                type: 'UserNameChanged',
                userId: 'user-1',
            };

            // Act
            const result = (await subscribe(
                schema,
                parse(subscription),
                {},
                mockContext,
                { topic: 'sync' }
            )) as AsyncIterableIterator<ExecutionResult<ISyncEvent>>;

            // Uncomment to verify message is published
            // await pubSub.subscribe('sync', (msg) => {
            //     console.log(msg);
            // });

            await pubSub.publish('sync', event);

            // *** TEST FAILS WITH TIMEOUT HERE ***
            await expect(result.next()).resolves.toBe({ id: 'foo' });
        });
    });
});

[robinelvin]

Further to the above, if I add a second test this sometimes passes and I see the log of the subscription resolver for the event. Adding a third test also sometimes passes but I only ever see 2 events and not the 3 emitted. The tests are intermittent and often time out on repeated runs even if the subscription resolver is fired. I'm not sure if this is my tests, TypeGraphQL, or something to do with the Redis iterator.