-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #1902 from alkem-io/server-1929
Subscriptions for Aspects
- Loading branch information
Showing
10 changed files
with
230 additions
and
13 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
69 changes: 69 additions & 0 deletions
69
src/core/microservices/subscription.context.aspect.created.factory.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| import { ConfigurationTypes, LogContext } from '@common/enums'; | ||
| import { LoggerService } from '@nestjs/common'; | ||
| import { ConfigService } from '@nestjs/config'; | ||
| import { AMQPPubSub } from 'graphql-amqp-subscriptions'; | ||
| import { PubSubEngine } from 'graphql-subscriptions'; | ||
| import amqp from 'amqplib'; | ||
| import { MessagingQueue } from '@common/enums/messaging.queue'; | ||
|
|
||
| export async function subscriptionContextAspectCreatedFactory( | ||
| logger: LoggerService, | ||
| configService: ConfigService | ||
| ): Promise<PubSubEngine | undefined> { | ||
| const rabbitMqOptions = configService.get( | ||
| ConfigurationTypes.MICROSERVICES | ||
| )?.rabbitmq; | ||
| const connectionOptions = rabbitMqOptions.connection; | ||
| const connectionString = `amqp://${connectionOptions.user}:${connectionOptions.password}@${connectionOptions.host}:${connectionOptions.port}?heartbeat=30`; | ||
|
|
||
| return amqp | ||
| .connect(connectionString) | ||
| .then(conn => { | ||
| return new AMQPPubSub({ | ||
| connection: conn, | ||
| exchange: { | ||
| // RabbitMQ subscriptions exchange name | ||
| name: 'alkemio-graphql-subscriptions', | ||
| // RabbitMQ exchange type. There are 4 exchange types: | ||
| // TOPIC - Topic exchanges route messages to one or many queues based on matching between a message routing key and the pattern that was used to bind a queue to an exchange. | ||
| // The topic exchange type is often used to implement various publish/subscribe pattern variations. Topic exchanges are commonly used for the multicast routing of messages. | ||
| // DIRECT - A direct exchange delivers messages to queues based on the message routing key. | ||
| // A direct exchange is ideal for the unicast routing of messages (although they can be used for multicast routing as well). | ||
| // HEADERS - A headers exchange is designed for routing on multiple attributes that are more easily expressed as message headers than a routing key. Headers exchanges ignore the routing key attribute. | ||
| // Instead, the attributes used for routing are taken from the headers attribute. A message is considered matching if the value of the header equals the value specified upon binding. | ||
| // FANOUT - A fanout exchange routes messages to all of the queues that are bound to it and the routing key is ignored. | ||
| // If N queues are bound to a fanout exchange, when a new message is published to that exchange a copy of the message is delivered to all N queues. | ||
| // Fanout exchanges are ideal for the broadcast routing of messages. | ||
| type: 'topic', | ||
| options: { | ||
| // the exchange will survive a broker restart | ||
| durable: true, | ||
| // exchange is deleted when last queue is unbound from it | ||
| autoDelete: false, | ||
| }, | ||
| }, | ||
| queue: { | ||
| name: MessagingQueue.SUBSCRIPTION_CONTEXT_ASPECT_CREATED, | ||
| options: { | ||
| // used by only one connection and the queue will be deleted when that connection closes | ||
| exclusive: false, | ||
| // the queue will survive a broker restart | ||
| durable: true, | ||
| // queue that has had at least one consumer is deleted when last consumer unsubscribes | ||
| autoDelete: false, | ||
| }, | ||
| // Unbind from the RabbitMQ queue when disposing the pubsub connection | ||
| unbindOnDispose: false, | ||
| // Delete the RabbitMQ queue when disposing the pubsub connection | ||
| deleteOnDispose: false, | ||
| }, | ||
| }); | ||
| }) | ||
| .catch(err => { | ||
| logger.error( | ||
| `Could not connect to RabbitMQ: ${err}, logging in...`, | ||
| LogContext.SUBSCRIPTIONS | ||
| ); | ||
| return undefined; | ||
| }); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
95 changes: 95 additions & 0 deletions
95
src/domain/context/context/context.resolver.subscriptions.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,95 @@ | ||
| import { CurrentUser } from '@common/decorators/current-user.decorator'; | ||
| import { SubscriptionType } from '@common/enums/subscription.type'; | ||
| import { AgentInfo } from '@core/authentication/agent-info'; | ||
| import { GraphqlGuard } from '@core/authorization'; | ||
| import { Inject, LoggerService, UseGuards } from '@nestjs/common'; | ||
| import { Args, Resolver, Subscription } from '@nestjs/graphql'; | ||
| import { WINSTON_MODULE_NEST_PROVIDER } from 'nest-winston'; | ||
| import { PubSubEngine } from 'graphql-subscriptions'; | ||
| import { LogContext } from '@common/enums/logging.context'; | ||
| import { UUID } from '@domain/common/scalars/scalar.uuid'; | ||
| import { AuthorizationService } from '@core/authorization/authorization.service'; | ||
| import { AuthorizationPrivilege } from '@common/enums/authorization.privilege'; | ||
| import { SUBSCRIPTION_CONTEXT_ASPECT_CREATED } from '@common/constants/providers'; | ||
| import { ContextService } from '@domain/context/context/context.service'; | ||
| import { ContextAspectCreated } from '@domain/context/context/dto/context.dto.event.aspect.created'; | ||
|
|
||
| @Resolver() | ||
| export class ContextResolverSubscriptions { | ||
| constructor( | ||
| @Inject(WINSTON_MODULE_NEST_PROVIDER) | ||
| private readonly logger: LoggerService, | ||
| @Inject(SUBSCRIPTION_CONTEXT_ASPECT_CREATED) | ||
| private subscriptionAspectCreated: PubSubEngine, | ||
| private contextService: ContextService, | ||
| private authorizationService: AuthorizationService | ||
| ) {} | ||
|
|
||
| @UseGuards(GraphqlGuard) | ||
| @Subscription(() => ContextAspectCreated, { | ||
| description: | ||
| 'Receive new Update messages on Communities the currently authenticated User is a member of.', | ||
| async resolve( | ||
| this: ContextResolverSubscriptions, | ||
| value: ContextAspectCreated, | ||
| _: unknown, | ||
| context: { req: { user: AgentInfo } } | ||
| ): Promise<ContextAspectCreated> { | ||
| const agentInfo = context.req.user; | ||
| const logMsgPrefix = `[User (${agentInfo.email}) Context Aspects] - `; | ||
| this.logger.verbose?.( | ||
| `${logMsgPrefix} sending out event for Aspects on Context: ${value.contextID} `, | ||
| LogContext.SUBSCRIPTIONS | ||
| ); | ||
| return value; | ||
| }, | ||
| async filter( | ||
| this: ContextResolverSubscriptions, | ||
| payload: ContextAspectCreated, | ||
| variables: { contextID: string }, | ||
| context: { req: { user: AgentInfo } } | ||
| ) { | ||
| const agentInfo = context.req.user; | ||
| const logMsgPrefix = `[User (${agentInfo.email}) Context Aspects] - `; | ||
| this.logger.verbose?.( | ||
| `${logMsgPrefix} Filtering event '${payload.eventID}'`, | ||
| LogContext.SUBSCRIPTIONS | ||
| ); | ||
|
|
||
| const isSameContext = payload.contextID === variables.contextID; | ||
| this.logger.verbose?.( | ||
| `${logMsgPrefix} Filter result is ${isSameContext}`, | ||
| LogContext.SUBSCRIPTIONS | ||
| ); | ||
| return isSameContext; | ||
| }, | ||
| }) | ||
| async contextAspectCreated( | ||
| @CurrentUser() agentInfo: AgentInfo, | ||
| @Args({ | ||
| name: 'contextID', | ||
| type: () => UUID, | ||
| description: 'The ID of the Context to subscribe to.', | ||
| nullable: false, | ||
| }) | ||
| contextID: string | ||
| ) { | ||
| const logMsgPrefix = `[User (${agentInfo.email}) Context Aspects] - `; | ||
| this.logger.verbose?.( | ||
| `${logMsgPrefix} Subscribing to the following Context Aspects: ${contextID}`, | ||
| LogContext.SUBSCRIPTIONS | ||
| ); | ||
| // check the user has the READ privilege | ||
| const context = await this.contextService.getContextOrFail(contextID); | ||
| await this.authorizationService.grantAccessOrFail( | ||
| agentInfo, | ||
| context.authorization, | ||
| AuthorizationPrivilege.READ, | ||
| `subscription to new Aspects on Context: ${context.id}` | ||
| ); | ||
|
|
||
| return this.subscriptionAspectCreated.asyncIterator( | ||
| SubscriptionType.CONTEXT_ASPECT_CREATED | ||
| ); | ||
| } | ||
| } |
20 changes: 20 additions & 0 deletions
20
src/domain/context/context/dto/context.dto.event.aspect.created.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,20 @@ | ||
| import { IAspect } from '@src/domain'; | ||
| import { Field, ObjectType } from '@nestjs/graphql'; | ||
|
|
||
| @ObjectType('ContextAspectCreated') | ||
| export class ContextAspectCreated { | ||
| eventID!: string; | ||
|
|
||
| @Field(() => String, { | ||
| nullable: false, | ||
| description: | ||
| 'The identifier for the Context on which the aspect was created.', | ||
| }) | ||
| contextID!: string; | ||
|
|
||
| @Field(() => IAspect, { | ||
| nullable: false, | ||
| description: 'The aspect that has been created.', | ||
| }) | ||
| aspect!: IAspect; | ||
| } |