Přeskočit na hlavní obsah

The 1st method - gRPC Message Consumer

Overview​

  • This is the first way of communication:
    • To register consumer and consume events of application domain services using message gateway
    • gRPC API is defined by gRPC protocol, described by a shared Protocol Buffers (.proto) contract
    • The event messages are supported by .NET connectors of platform services using nuget packages
    • Any other gRPC-capable technology (Java, Go, ...) can generate its own client directly from the same .proto contract, without depending on the .NET package

Message Gateway​

MessageGateway is platform service encapsulating external communication with message broker to HTTP based technologies, gRPC and Webhooks. It is the only way for external cloud and on-premise services integrated with AVAplace platform.

See more: OIDC/OAuth2 authentication

How gRPC communication works​

The gRPC channel between a consumer and the MessageGateway is a single, long-lived bidirectional streaming RPC (MessageGatewayConsumerService.MessageStream). The consumer opens this full-duplex stream once and keeps it open - unlike REST or Webhook, there is no separate request/response per message.

Once the stream is open:

  • the gateway pushes a ReceivedMessage to the consumer whenever a message matching its registered consumer/message type is routed to it,
  • the consumer replies on the same stream with a MessageDelivery acknowledgement for each message it has processed.

Each received message carries:

  • a unique message id, used to correlate it with the consumer's acknowledgement,
  • the routing message type (the same value used during consumer registration),
  • headers as key/value pairs,
  • the serialized payload (typically JSON),
  • the time the gateway sent the message,
  • optionally, the fully qualified contract type name, used by the client to deserialize the payload into a strongly typed object.

Each delivery acknowledgement carries the message id and a delivery type:

delivery typemeaning
ACKthe message was processed successfully
NACKprocessing failed; the gateway will redeliver the message according to the reconnection/delivery policy

Because the gRPC service is described by a standard Protocol Buffers contract, it isn't tied to .NET. The ASOL.MessageGateway.Connector.gRPC.Consumer nuget package is itself built on this contract and provides ready-to-use consumer registration and message dispatch for .NET applications (see below). Any other gRPC-capable stack - Java, Go, etc. - can generate its own client straight from the .proto contract using standard protoc/grpc tooling, without referencing the .NET package.

syntax = "proto3";

package asol.messagegateway.v1;

option csharp_namespace = "ASOL.MessageGateway.V1";
option java_package = "asol.messagegateway.v1";
option java_outer_classname = "AsolMessageGatewayConsumerProto";
option go_package = "go.asol.io/proto/messagegateway/v1";


service MessageGatewayConsumerService {
  rpc MessageStream(stream MessageDelivery) returns (stream ReceivedMessage); 
}

message MessageDelivery {
  string message_id = 1;
  DeliveryType delivery_type = 2;
}

message ReceivedMessage {
  string message_id = 1;
  string message_type = 2;
  map<string, string> headers = 3;
  string body = 4;
  string sent_time = 5;
  optional string contract_type = 6;
}

enum DeliveryType {
  ACK = 0;
  NACK = 1;
}

Message headers​

Each message delivered on the stream carries context information as entries of the headers map field on ReceivedMessage (map<string, string>, defined in asol_message_gateway_consumer.proto).

Headers such as X-Tenant-Id and X-UserClaim-* describe the tenant-context and user-context of the specific notification message. These values are created by the originating platform service when that service publishes the notification message to the message bus. They are propagated with the message and later included by MessageGateway into the gRPC request.

headerdescription
X-UserClaim-ActortPerson id (Actor) in user context.
X-UserClaim-client_idClient identification in user or service context.
X-UserClaim-issIssuer in user or service context.
X-UserClaim-localeLanguage and region context (RFC 5646 / ISO-639-2) in user or service context.
X-UserClaim-orgs_codesSelected organizations (organization.code) - Organization national number|Country code.
X-UserClaim-subUser identification (SSO unique identifier) in user context.
X-UserClaim-tidTenant identification of selected tenant in user or service context.
X-UserClaimsExtendedExtended claims in user or service context in a plain JSON array of key/value pairs.
X-MessageSecurityCodeThe legacy message security code for the message to keep compatibility with legacy systems.
X-Tenant-IdTenant context identification of the specific message. The value originates in the service that published the message.

A gRPC consumer can therefore need the tenant header to process the notification in the correct tenant context.

  • X-UserClaim-tid - represents the specific tenant-context in user identity (so called runtime-context)
  • X-Tenant-Id - represents the specific tenant-context explicitly assigned to message (so called tenant-context), typically used when user identity is impersonated by service context

Under normal circumstances, both values should be the same.

The message-context headers can also be used by the messaging infrastructure for message filtering/routing before the notification is delivered. On the consumer side they describe the original message context and can be useful for processing, diagnostics, troubleshooting and correlation according to the integration contract.

X-MessageSecurityCode is a legacy proprietary mechanism that was introduced to protect propagated messages against spoofing or modification. It is still present in the current webhook contract and legacy integrations may be required to validate it, but it is no longer recommended for new integrations.

Consumers implemented with the official ASOL.MessageGateway.Connector.gRPC.Consumer .NET client library do not need to parse the headers map themselves — the library already maps these entries into a ClaimsPrincipal and tenant context per message. Consumers implemented in other languages (Java, Go, Node.js, Python, …), using the plain gRPC stubs generated from asol_message_gateway_consumer.proto, receive ReceivedMessage.headers as an ordinary string map and must read/interpret the entries above themselves.

Before the notification payload is accepted for business processing, the consumer should:

  1. Connect to MessageGateway only over a TLS-secured gRPC channel.
  2. Obtain a valid access token via OAuth2 client-credentials from the AVAplace identity-provider and send it as Authorization: Bearer <token> metadata when opening the stream.
  3. Keep the token refreshed for the lifetime of the connection, requesting a new one before the current one expires.
  4. Read the message-specific user identity and tenant context from the headers map of each ReceivedMessage, as required by the integration contract.
  5. Validate the message's message_type and contract_type fields and payload required by the integration before business processing.
  6. Use message_id to make processing idempotent. A repeated delivery of the same message must not create the same business effect twice.
  7. Send ACK (or NACK on failure) as a MessageDelivery on the request stream only after the notification has been accepted according to the consumer's processing strategy. A NACK, or the message's delivery-TTL expiring without a response, triggers redelivery.

The exact implementation depends on the consumer technology. A standard OAuth2/OIDC client-credentials library is recommended, since it handles token acquisition, refresh and error handling around signing-key rotation on the identity-provider side. See the technical gRPC integration examples for implementation guidance.

How to consume gRPC message using message gateway​

Consume domain entity released event using message gateway​

  • Example - how to consume gRPC message(s):

Registration example (Startup.cs):

using ASOL.MessageGateway.Connector.gRPC.Consumer.Extensions;

services.AddMessageGatewayGrpcConsumer(configuration, configurator =>
{
    configurator.Consumer<Example1_MyEntityReleasedEventConsumer>();
    // ... other consumers ...
});

Usage example:

using ASOL.Core.Identity;
using ASOL.MessageGateway.Connector.gRPC.Consumer;

private readonly IRuntimeContext _runtimeContext;

/// <summary>
/// Example of common domain event consumer against platform message gateway.
/// This example doesn't contain real implementation, it's only for demonstration purposes.
/// </summary>
[ConsumerDefinition("ASOL.Example1.MyEntityReleased")]
public async Task Consume(Example1_MyEntityReleased myEntityReleased)
{
    //(optional) check tenant context of received message (exception is thrown if not available)
    _ = _runtimeContext.Security.TenantId;

    //(optional) check user context of received message (exception is thrown if not available)
    _ = _runtimeContext.Security.UserId;

    //... evaluate if event should be processed or ignored

    // *** [authenticate +] call endpoint ***
    //... get details from domain service or persistence storage using MyEntity identifier

    //... process delivered message with retrieved payload
}

Configuration changes of gRPC message consumer​

The ASOL.MessageGateway.Connector.gRPC.Consumer nuget package (version 0.0.1.49076-dev) contains breaking changes in MessageGatewayGrpcConsumerClientOptions.
It is strongly recommended to upgrade package to version 0.0.1.52050-dev or 0.0.1.51809-dev at least. Both versions contain the same fixes and differ only in the versions of their dependent libraries. Please choose the one that best fits your current requirements.

Important: Please, switch from the single BaseUrl property to the two separate properties GateBaseUrl and ApiBaseUrl, and start using the new gate endpoints as soon as possible across all stages, including BETA, DEMO and PROD. Support for legacy gRPC communication URLs has been removed on development stages, except PROD stage. In the production environment, legacy URL will remain functional without changes until it is confirmed that they are no longer in use.

Note: legacy URL provides unstable connections due mix of used technologies. Use separate URLs for gRPC and REST API to avoid connection issues.

The communication of REST API and gRPC channel was split into two base URLs due their different requirements:

  • GateBaseUrl - base URL for gRPC full-duplex channel
    • BETA stage: https://beta-gate.avaplace.com
    • DEMO stage: https://demo-gate.avaplace.com
    • PROD stage: https://prod-gate.avaplace.com
  • ApiBaseUrl - base URL for REST API http(s) calls
    • BETA stage: https://beta.avaplace.com/api/asol/msggw
    • DEMO stage: https://demo.avaplace.com/api/asol/msggw
    • PROD stage: https://avaplace.com/api/asol/msggw

Usage example - appsettings.json:

  "MessageGatewayGrpcConsumerClientOptions": {
    "GateBaseUrl": "https://demo-gate.avaplace.com",          // PROD stage: "https://prod-gate.avaplace.com"
    "ApiBaseUrl": "https://demo.avaplace.com/api/asol/msggw", // PROD stage: "https://avaplace.com/api/asol/msggw"
    "ConsumerCode": "<place your consumerCode here>"
  },

See examples: