nodejs/src/producer/Producer.ts - rocketmq-clients - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 import assert from 'node:assert';
 import { setTimeout } from 'node:timers/promises';
 import {
   ClientType,
   MessageType,
   TransactionResolution,
 } from '../../proto/apache/rocketmq/v2/definition_pb';
 import {
   EndTransactionRequest,
   HeartbeatRequest,
   NotifyClientTerminationRequest,
   RecoverOrphanedTransactionCommand,
   SendMessageRequest,
 } from '../../proto/apache/rocketmq/v2/service_pb';
 import {
   Endpoints,
   MessageQueue,
   TopicRouteData,
 } from '../route';
 import {
   ExponentialBackoffRetryPolicy,
 } from '../retry';
 import { StatusChecker, TooManyRequestsException } from '../exception';
 import { BaseClient, BaseClientOptions, Settings } from '../client';
 import { PublishingMessage, MessageOptions, MessageView, Message } from '../message';
 import { PublishingSettings } from './PublishingSettings';
 import { TransactionChecker } from './TransactionChecker';
 import { PublishingLoadBalancer } from './PublishingLoadBalancer';
 import { SendReceipt } from './SendReceipt';
 import { Transaction } from './Transaction';
 import { createResource } from '../util';

 export interface ProducerOptions extends BaseClientOptions {
   topic?: string | string[];
   maxAttempts?: number;
   checker?: TransactionChecker;
 }

 export class Producer extends BaseClient {
   #publishingSettings: PublishingSettings;
   #checker?: TransactionChecker;
   #publishingRouteDataCache = new Map<string, PublishingLoadBalancer>();

   constructor(options: ProducerOptions) {
     if (!options.topics && options.topic) {
       options.topics = Array.isArray(options.topic) ? options.topic : [ options.topic ];
     }
     super(options);
     // https://rocketmq.apache.org/docs/introduction/03limits/
     // Default max number of message sending retries is 3
     const retryPolicy = ExponentialBackoffRetryPolicy.immediatelyRetryPolicy(options.maxAttempts ?? 3);
     this.#publishingSettings = new PublishingSettings(options.namespace, this.clientId, this.endpoints, retryPolicy,
       this.requestTimeout, this.topics);
     this.#checker = options.checker;
   }

   get publishingSettings() {
     return this.#publishingSettings;
   }

   beginTransaction() {
     assert(this.#checker, 'Transaction checker should not be null');
     return new Transaction(this);
   }

   async endTransaction(endpoints: Endpoints, message: Message, messageId: string,
     transactionId: string, resolution: TransactionResolution) {
     const request = new EndTransactionRequest()
       .setMessageId(messageId)
       .setTransactionId(transactionId)
       .setTopic(createResource(message.topic).setResourceNamespace(this.namespace))
       .setResolution(resolution);
     const response = await this.rpcClientManager.endTransaction(endpoints, request, this.requestTimeout);
     StatusChecker.check(response.getStatus()?.toObject());
   }

   async onRecoverOrphanedTransactionCommand(endpoints: Endpoints, command: RecoverOrphanedTransactionCommand) {
     const transactionId = command.getTransactionId();
     const messagePB = command.getMessage()!;
     const messageId = messagePB.getSystemProperties()!.getMessageId();
     if (!this.#checker) {
       this.logger.error('No transaction checker registered, ignore it, messageId=%s, transactionId=%s, endpoints=%s, clientId=%s',
         messageId, transactionId, endpoints, this.clientId);
       return;
     }
     let messageView: MessageView;
     try {
       messageView = new MessageView(messagePB);
     } catch (err) {
       this.logger.error('[Bug] Failed to decode message during orphaned transaction message recovery, messageId=%s, transactionId=%s, endpoints=%s, clientId=%s, error=%s',
         messageId, transactionId, endpoints, this.clientId, err);
       return;
     }

     try {
       const resolution = await this.#checker.check(messageView);
       if (resolution === null || resolution === TransactionResolution.TRANSACTION_RESOLUTION_UNSPECIFIED) {
         return;
       }
       await this.endTransaction(endpoints, messageView, messageId, transactionId, resolution);
       this.logger.info('Recover orphaned transaction message success, transactionId=%s, resolution=%s, messageId=%s, clientId=%s',
         transactionId, resolution, messageId, this.clientId);
     } catch (err) {
       this.logger.error('Exception raised while checking the transaction, messageId=%s, transactionId=%s, endpoints=%s, clientId=%s, error=%s',
         messageId, transactionId, endpoints, this.clientId, err);
       return;
     }
   }

   protected getSettings(): Settings {
     return this.#publishingSettings;
   }

   protected wrapHeartbeatRequest(): HeartbeatRequest {
     return new HeartbeatRequest()
       .setClientType(ClientType.PRODUCER);
   }

   protected wrapNotifyClientTerminationRequest(): NotifyClientTerminationRequest {
     return new NotifyClientTerminationRequest();
   }

   async send(message: MessageOptions, transaction?: Transaction) {
     if (!transaction) {
       const sendReceipts = await this.#send([ message ], false);
       return sendReceipts[0];
     }

     const publishingMessage = transaction.tryAddMessage(message);
     const sendReceipts = await this.#send([ message ], true);
     const sendReceipt = sendReceipts[0];
     transaction.tryAddReceipt(publishingMessage, sendReceipt);
     return sendReceipt;
   }

   async #send(messages: MessageOptions[], txEnabled: boolean) {
     const pubMessages: PublishingMessage[] = [];
     const topics = new Set<string>();
     for (const message of messages) {
       pubMessages.push(new PublishingMessage(message, this.#publishingSettings, txEnabled));
       topics.add(message.topic);
     }
     if (topics.size > 1) {
       throw new TypeError(`Messages to send have different topics=${JSON.stringify(topics)}`);
     }
     const topic = pubMessages[0].topic;
     const messageType = pubMessages[0].messageType;
     const messageGroup = pubMessages[0].messageGroup;
     const messageTypes = new Set(pubMessages.map(m => m.messageType));
     if (messageTypes.size > 1) {
       throw new TypeError(`Messages to send have different types=${JSON.stringify(messageTypes)}`);
     }

     // Message group must be same if message type is FIFO, or no need to proceed.
     if (messageType === MessageType.FIFO) {
       const messageGroups = new Set(pubMessages.map(m => m.messageGroup!));
       if (messageGroups.size > 1) {
         throw new TypeError(`FIFO messages to send have message groups, messageGroups=${JSON.stringify(messageGroups)}`);
       }
     }

     // Get publishing topic route.
     const loadBalancer = await this.#getPublishingLoadBalancer(topic);
     // Prepare the candidate message queue(s) for retry-sending in advance.
     const candidates = messageGroup ? [ loadBalancer.takeMessageQueueByMessageGroup(messageGroup) ] :
       this.#takeMessageQueues(loadBalancer);
     return await this.#send0(topic, messageType, candidates, pubMessages, 1);
   }

   #wrapSendMessageRequest(pubMessages: PublishingMessage[], mq: MessageQueue) {
     const request = new SendMessageRequest();
     for (const pubMessage of pubMessages) {
       if (this.namespace) {
         request.addMessages(pubMessage.toProtobuf(this.namespace, mq));
       } else {
         request.addMessages(pubMessage.toProtobuf('', mq));
       }
     }
     return request;
   }

   /**
    * Isolate specified Endpoints
    */
   #isolate(endpoints: Endpoints) {
     this.isolated.set(endpoints.facade, endpoints);
   }

   async #send0(topic: string, messageType: MessageType, candidates: MessageQueue[],
     messages: PublishingMessage[], attempt: number): Promise<SendReceipt[]> {
     // Calculate the current message queue.
     const index = (attempt - 1) % candidates.length;
     const mq = candidates[index];
     const acceptMessageTypes = mq.acceptMessageTypesList;
     if (this.#publishingSettings.isValidateMessageType() && !acceptMessageTypes.includes(messageType)) {
       throw new TypeError('Current message type not match with ' +
         'topic accept message types, topic=' + topic + ', actualMessageType=' + messageType + ', ' +
         'acceptMessageTypes=' + JSON.stringify(acceptMessageTypes));
     }
     const endpoints = mq.broker.endpoints;
     const maxAttempts = this.#getRetryPolicy().getMaxAttempts();
     const request = this.#wrapSendMessageRequest(messages, mq);
     let sendReceipts: SendReceipt[] = [];
     try {
       const response = await this.rpcClientManager.sendMessage(endpoints, request, this.requestTimeout);
       sendReceipts = SendReceipt.processResponseInvocation(mq, response);
     } catch (err) {
       const messageIds = messages.map(m => m.messageId);
       // Isolate endpoints because of sending failure.
       this.#isolate(endpoints);
       if (attempt >= maxAttempts) {
         // No need more attempts.
         this.logger.error('Failed to send message(s) finally, run out of attempt times, maxAttempts=%s, attempt=%s, topic=%s, messageId(s)=%s, endpoints=%s, clientId=%s, error=%s',
           maxAttempts, attempt, topic, messageIds, endpoints, this.clientId, err);
         throw err;
       }
       // No need more attempts for transactional message.
       if (messageType === MessageType.TRANSACTION) {
         this.logger.error('Failed to send transactional message finally, maxAttempts=%s, attempt=%s, topic=%s, messageId(s)=%s, endpoints=%s, clientId=%s, error=%s',
           maxAttempts, attempt, topic, messageIds, endpoints, this.clientId, err);
         throw err;
       }
       // Try to do more attempts.
       const nextAttempt = 1 + attempt;
       // Retry immediately if the request is not throttled.
       if (!(err instanceof TooManyRequestsException)) {
         this.logger.warn('Failed to send message, would attempt to resend right now, maxAttempts=%s, attempt=%s, topic=%s, messageId(s)=%s, endpoints=%s, clientId=%s, error=%s',
           maxAttempts, attempt, topic, messageIds, endpoints, this.clientId, err);
         return this.#send0(topic, messageType, candidates, messages, nextAttempt);
       }
       const delay = this.#getRetryPolicy().getNextAttemptDelay(nextAttempt);
       this.logger.warn('Failed to send message due to too many requests, would attempt to resend after %sms, maxAttempts=%s, attempt=%s, topic=%s, messageId(s)=%s, endpoints=%s, clientId=%s, error=%s',
         delay, maxAttempts, attempt, topic, messageIds, endpoints, this.clientId, err);
       await setTimeout(delay);
       return this.#send0(topic, messageType, candidates, messages, nextAttempt);
     }

     // Resend message(s) successfully.
     if (attempt > 1) {
       const messageIds = sendReceipts.map(r => r.messageId);
       this.logger.info('Resend message successfully, topic=%s, messageId(s)=%j, maxAttempts=%s, attempt=%s, endpoints=%s, clientId=%s',
         topic, messageIds, maxAttempts, attempt, endpoints, this.clientId);
     }
     // Send message(s) successfully on first attempt, return directly.
     return sendReceipts;
   }

   async #getPublishingLoadBalancer(topic: string) {
     let loadBalancer = this.#publishingRouteDataCache.get(topic);
     if (!loadBalancer) {
       const topicRouteData = await this.getRouteData(topic);
       loadBalancer = this.#updatePublishingLoadBalancer(topic, topicRouteData);
     }
     return loadBalancer;
   }

   #updatePublishingLoadBalancer(topic: string, topicRouteData: TopicRouteData) {
     let loadBalancer = this.#publishingRouteDataCache.get(topic);
     if (loadBalancer) {
       loadBalancer = loadBalancer.update(topicRouteData);
     } else {
       loadBalancer = new PublishingLoadBalancer(topicRouteData);
     }
     this.#publishingRouteDataCache.set(topic, loadBalancer);
     return loadBalancer;
   }

   /**
    * Take message queue(s) from route for message publishing.
    */
   #takeMessageQueues(loadBalancer: PublishingLoadBalancer) {
     return loadBalancer.takeMessageQueues(this.isolated, this.#getRetryPolicy().getMaxAttempts());
   }

   #getRetryPolicy() {
     return this.#publishingSettings.getRetryPolicy()!;
   }
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	import assert from 'node:assert';
	import { setTimeout } from 'node:timers/promises';
	import {
	ClientType,
	MessageType,
	TransactionResolution,
	} from '../../proto/apache/rocketmq/v2/definition_pb';
	import {
	EndTransactionRequest,
	HeartbeatRequest,
	NotifyClientTerminationRequest,
	RecoverOrphanedTransactionCommand,
	SendMessageRequest,
	} from '../../proto/apache/rocketmq/v2/service_pb';
	import {
	Endpoints,
	MessageQueue,
	TopicRouteData,
	} from '../route';
	import {
	ExponentialBackoffRetryPolicy,
	} from '../retry';
	import { StatusChecker, TooManyRequestsException } from '../exception';
	import { BaseClient, BaseClientOptions, Settings } from '../client';
	import { PublishingMessage, MessageOptions, MessageView, Message } from '../message';
	import { PublishingSettings } from './PublishingSettings';
	import { TransactionChecker } from './TransactionChecker';
	import { PublishingLoadBalancer } from './PublishingLoadBalancer';
	import { SendReceipt } from './SendReceipt';
	import { Transaction } from './Transaction';
	import { createResource } from '../util';

	export interface ProducerOptions extends BaseClientOptions {
	topic?: string \| string[];
	maxAttempts?: number;
	checker?: TransactionChecker;
	}

	export class Producer extends BaseClient {
	#publishingSettings: PublishingSettings;
	#checker?: TransactionChecker;
	#publishingRouteDataCache = new Map<string, PublishingLoadBalancer>();

	constructor(options: ProducerOptions) {
	if (!options.topics && options.topic) {
	options.topics = Array.isArray(options.topic) ? options.topic : [ options.topic ];
	}
	super(options);
	// https://rocketmq.apache.org/docs/introduction/03limits/
	// Default max number of message sending retries is 3
	const retryPolicy = ExponentialBackoffRetryPolicy.immediatelyRetryPolicy(options.maxAttempts ?? 3);
	this.#publishingSettings = new PublishingSettings(options.namespace, this.clientId, this.endpoints, retryPolicy,
	this.requestTimeout, this.topics);
	this.#checker = options.checker;
	}

	get publishingSettings() {
	return this.#publishingSettings;
	}

	beginTransaction() {
	assert(this.#checker, 'Transaction checker should not be null');
	return new Transaction(this);
	}

	async endTransaction(endpoints: Endpoints, message: Message, messageId: string,
	transactionId: string, resolution: TransactionResolution) {
	const request = new EndTransactionRequest()
	.setMessageId(messageId)
	.setTransactionId(transactionId)
	.setTopic(createResource(message.topic).setResourceNamespace(this.namespace))
	.setResolution(resolution);
	const response = await this.rpcClientManager.endTransaction(endpoints, request, this.requestTimeout);
	StatusChecker.check(response.getStatus()?.toObject());
	}

	async onRecoverOrphanedTransactionCommand(endpoints: Endpoints, command: RecoverOrphanedTransactionCommand) {
	const transactionId = command.getTransactionId();
	const messagePB = command.getMessage()!;
	const messageId = messagePB.getSystemProperties()!.getMessageId();
	if (!this.#checker) {
	this.logger.error('No transaction checker registered, ignore it, messageId=%s, transactionId=%s, endpoints=%s, clientId=%s',
	messageId, transactionId, endpoints, this.clientId);
	return;
	}
	let messageView: MessageView;
	try {
	messageView = new MessageView(messagePB);
	} catch (err) {
	this.logger.error('[Bug] Failed to decode message during orphaned transaction message recovery, messageId=%s, transactionId=%s, endpoints=%s, clientId=%s, error=%s',
	messageId, transactionId, endpoints, this.clientId, err);
	return;
	}

	try {
	const resolution = await this.#checker.check(messageView);
	if (resolution === null \|\| resolution === TransactionResolution.TRANSACTION_RESOLUTION_UNSPECIFIED) {
	return;
	}
	await this.endTransaction(endpoints, messageView, messageId, transactionId, resolution);
	this.logger.info('Recover orphaned transaction message success, transactionId=%s, resolution=%s, messageId=%s, clientId=%s',
	transactionId, resolution, messageId, this.clientId);
	} catch (err) {
	this.logger.error('Exception raised while checking the transaction, messageId=%s, transactionId=%s, endpoints=%s, clientId=%s, error=%s',
	messageId, transactionId, endpoints, this.clientId, err);
	return;
	}
	}

	protected getSettings(): Settings {
	return this.#publishingSettings;
	}

	protected wrapHeartbeatRequest(): HeartbeatRequest {
	return new HeartbeatRequest()
	.setClientType(ClientType.PRODUCER);
	}

	protected wrapNotifyClientTerminationRequest(): NotifyClientTerminationRequest {
	return new NotifyClientTerminationRequest();
	}

	async send(message: MessageOptions, transaction?: Transaction) {
	if (!transaction) {
	const sendReceipts = await this.#send([ message ], false);
	return sendReceipts[0];
	}

	const publishingMessage = transaction.tryAddMessage(message);
	const sendReceipts = await this.#send([ message ], true);
	const sendReceipt = sendReceipts[0];
	transaction.tryAddReceipt(publishingMessage, sendReceipt);
	return sendReceipt;
	}

	async #send(messages: MessageOptions[], txEnabled: boolean) {
	const pubMessages: PublishingMessage[] = [];
	const topics = new Set<string>();
	for (const message of messages) {
	pubMessages.push(new PublishingMessage(message, this.#publishingSettings, txEnabled));
	topics.add(message.topic);
	}
	if (topics.size > 1) {
	throw new TypeError(`Messages to send have different topics=${JSON.stringify(topics)}`);
	}
	const topic = pubMessages[0].topic;
	const messageType = pubMessages[0].messageType;
	const messageGroup = pubMessages[0].messageGroup;
	const messageTypes = new Set(pubMessages.map(m => m.messageType));
	if (messageTypes.size > 1) {
	throw new TypeError(`Messages to send have different types=${JSON.stringify(messageTypes)}`);
	}

	// Message group must be same if message type is FIFO, or no need to proceed.
	if (messageType === MessageType.FIFO) {
	const messageGroups = new Set(pubMessages.map(m => m.messageGroup!));
	if (messageGroups.size > 1) {
	throw new TypeError(`FIFO messages to send have message groups, messageGroups=${JSON.stringify(messageGroups)}`);
	}
	}

	// Get publishing topic route.
	const loadBalancer = await this.#getPublishingLoadBalancer(topic);
	// Prepare the candidate message queue(s) for retry-sending in advance.
	const candidates = messageGroup ? [ loadBalancer.takeMessageQueueByMessageGroup(messageGroup) ] :
	this.#takeMessageQueues(loadBalancer);
	return await this.#send0(topic, messageType, candidates, pubMessages, 1);
	}

	#wrapSendMessageRequest(pubMessages: PublishingMessage[], mq: MessageQueue) {
	const request = new SendMessageRequest();
	for (const pubMessage of pubMessages) {
	if (this.namespace) {
	request.addMessages(pubMessage.toProtobuf(this.namespace, mq));
	} else {
	request.addMessages(pubMessage.toProtobuf('', mq));
	}
	}
	return request;
	}

	/**
	* Isolate specified Endpoints
	*/
	#isolate(endpoints: Endpoints) {
	this.isolated.set(endpoints.facade, endpoints);
	}

	async #send0(topic: string, messageType: MessageType, candidates: MessageQueue[],
	messages: PublishingMessage[], attempt: number): Promise<SendReceipt[]> {
	// Calculate the current message queue.
	const index = (attempt - 1) % candidates.length;
	const mq = candidates[index];
	const acceptMessageTypes = mq.acceptMessageTypesList;
	if (this.#publishingSettings.isValidateMessageType() && !acceptMessageTypes.includes(messageType)) {
	throw new TypeError('Current message type not match with ' +
	'topic accept message types, topic=' + topic + ', actualMessageType=' + messageType + ', ' +
	'acceptMessageTypes=' + JSON.stringify(acceptMessageTypes));
	}
	const endpoints = mq.broker.endpoints;
	const maxAttempts = this.#getRetryPolicy().getMaxAttempts();
	const request = this.#wrapSendMessageRequest(messages, mq);
	let sendReceipts: SendReceipt[] = [];
	try {
	const response = await this.rpcClientManager.sendMessage(endpoints, request, this.requestTimeout);
	sendReceipts = SendReceipt.processResponseInvocation(mq, response);
	} catch (err) {
	const messageIds = messages.map(m => m.messageId);
	// Isolate endpoints because of sending failure.
	this.#isolate(endpoints);
	if (attempt >= maxAttempts) {
	// No need more attempts.
	this.logger.error('Failed to send message(s) finally, run out of attempt times, maxAttempts=%s, attempt=%s, topic=%s, messageId(s)=%s, endpoints=%s, clientId=%s, error=%s',
	maxAttempts, attempt, topic, messageIds, endpoints, this.clientId, err);
	throw err;
	}
	// No need more attempts for transactional message.
	if (messageType === MessageType.TRANSACTION) {
	this.logger.error('Failed to send transactional message finally, maxAttempts=%s, attempt=%s, topic=%s, messageId(s)=%s, endpoints=%s, clientId=%s, error=%s',
	maxAttempts, attempt, topic, messageIds, endpoints, this.clientId, err);
	throw err;
	}
	// Try to do more attempts.
	const nextAttempt = 1 + attempt;
	// Retry immediately if the request is not throttled.
	if (!(err instanceof TooManyRequestsException)) {
	this.logger.warn('Failed to send message, would attempt to resend right now, maxAttempts=%s, attempt=%s, topic=%s, messageId(s)=%s, endpoints=%s, clientId=%s, error=%s',
	maxAttempts, attempt, topic, messageIds, endpoints, this.clientId, err);
	return this.#send0(topic, messageType, candidates, messages, nextAttempt);
	}
	const delay = this.#getRetryPolicy().getNextAttemptDelay(nextAttempt);
	this.logger.warn('Failed to send message due to too many requests, would attempt to resend after %sms, maxAttempts=%s, attempt=%s, topic=%s, messageId(s)=%s, endpoints=%s, clientId=%s, error=%s',
	delay, maxAttempts, attempt, topic, messageIds, endpoints, this.clientId, err);
	await setTimeout(delay);
	return this.#send0(topic, messageType, candidates, messages, nextAttempt);
	}

	// Resend message(s) successfully.
	if (attempt > 1) {
	const messageIds = sendReceipts.map(r => r.messageId);
	this.logger.info('Resend message successfully, topic=%s, messageId(s)=%j, maxAttempts=%s, attempt=%s, endpoints=%s, clientId=%s',
	topic, messageIds, maxAttempts, attempt, endpoints, this.clientId);
	}
	// Send message(s) successfully on first attempt, return directly.
	return sendReceipts;
	}

	async #getPublishingLoadBalancer(topic: string) {
	let loadBalancer = this.#publishingRouteDataCache.get(topic);
	if (!loadBalancer) {
	const topicRouteData = await this.getRouteData(topic);
	loadBalancer = this.#updatePublishingLoadBalancer(topic, topicRouteData);
	}
	return loadBalancer;
	}

	#updatePublishingLoadBalancer(topic: string, topicRouteData: TopicRouteData) {
	let loadBalancer = this.#publishingRouteDataCache.get(topic);
	if (loadBalancer) {
	loadBalancer = loadBalancer.update(topicRouteData);
	} else {
	loadBalancer = new PublishingLoadBalancer(topicRouteData);
	}
	this.#publishingRouteDataCache.set(topic, loadBalancer);
	return loadBalancer;
	}

	/**
	* Take message queue(s) from route for message publishing.
	*/
	#takeMessageQueues(loadBalancer: PublishingLoadBalancer) {
	return loadBalancer.takeMessageQueues(this.isolated, this.#getRetryPolicy().getMaxAttempts());
	}

	#getRetryPolicy() {
	return this.#publishingSettings.getRetryPolicy()!;
	}
	}