Azure Core AMQP 用于AMQP操作的客户端库
提供不同功能的库 阿苏雷 使用 amqp 协议的 Javascript 库 . 其中一些共同功能包括:
- CBS 澳大利亚
- 通过AMQP发送请求和接收回复的请求-响应链接
- 不同的 AMQP 错误代码翻译出错 .
- 创建
ConnectionConfig通过解析连接字符串 - AMQP 信件头和信件属性转换
- 大多数AMQP基于Azure服务用于通信的AMQP消息数据转换(编码/解码)
- 重试不同操作的政策
开始
先决条件
- 节点.js版本: 6.x或更高.
- 我们会 还是鼓励你 以在任何特定时间安装最新的 LTS 版本 https://nodejs.org (英语)。 。 。 . 总是安装最新可用的LTS版本节点.js是一个好的做法.
- 正在安装节点.js on 窗口或macOS 使用可用的安装器 节点.js网站。如果您正在使用 基于 linux 的操作系统,然后可以发现容易遵循,一步安装指令 这里。 。 。 .
安装
- 安装此库
npm install @azure/amqp-common
rhea-promise是同级依赖。 您需要明确安装此库作为依赖 在您的申请.
IDE 开发环境
这个 sdk 是在 类型脚本 并且有很好的源代码文档. 强烈建议使用 vscode 密码 或任何能提供更好的内涵并暴露源代码文件的全部功率的IDE.
关键概念
这个 SDK 包含核心 AMQP 在使用 bu Azure SDKs 使用协议中的共同相关功能.
实例
请看一下 样本 详细样本目录. 您可以在样本.js文件中复制复制以下样本.
- git clone https://github.com/Azure/azure-sdk-for-js.git
- cd azure-sdk-for-js/sdk/core/amqp-common
- npm i
- npm i -g typescript #This is optional. However it is useful to have typescript installed globally on your box
- npm i -g ts-node #This is optional. However it is useful to have ts-node installed globally on your box
# Make sure to set the environment variables and then run
- ts-node ./samples/cbsAuth.ts
以下样本为EventHubs和Servicebus通用. 您可以找到事件Hub的具体样本, 输入 样本 目录.
例1 - CBS(基于请求的授权规格)
有关CBs授权的更多信息,请查看 这里。 。 。 .
注:下面的代码块后来被称作"./cbsAuth".
const { ConnectionContextBase, ConnectionConfig, CbsResponse } = require("@azure/amqp-common");
const dotenv = require("dotenv");
dotenv.config(); // Optional for loading environment configuration from a .env (config) file
export const str = process.env.CONNECTION_STRING || "";
export const path = process.env.ENTITY_PATH;
export const connectionConfig = ConnectionConfig.create(str, path);
const parameters = {
config: connectionConfig,
connectionProperties: {
product: "MSJSClient",
userAgent: "/js-amqp-common",
version: "0.1.0"
}
};
export const connectionContext = ConnectionContextBase.create(parameters);
/**
* audience The entity token audience in one of the following forms:
*
* - **ServiceBus**
* - **Sender**
* - `"sb://<yournamespace>.servicebus.windows.net/<queue-name>"`
* - `"sb://<yournamespace>.servicebus.windows.net/<topic-name>"`
*
* - **Receiver**
* - `"sb://<yournamespace>.servicebus.windows.net/<queue-name>"`
* - `"sb://<yournamespace>.servicebus.windows.net/<topic-name>"`
*
* - **ManagementClient**
* - `"sb://<your-namespace>.servicebus.windows.net/<queue-name>/$management"`.
* - `"sb://<your-namespace>.servicebus.windows.net/<topic-name>/$management"`.
*
* - **EventHubs**
* - **Sender**
* - `"sb://<yournamespace>.servicebus.windows.net/<hubName>"`
* - `"sb://<yournamespace>.servicebus.windows.net/<hubName>/Partitions/<partitionId>"`.
*
* - **Receiver**
* - `"sb://<your-namespace>.servicebus.windows.net/<event-hub-name>/ConsumerGroups/<consumer-group-name>/Partitions/<partition-id>"`.
*
* - **ManagementClient**
* - `"sb://<your-namespace>.servicebus.windows.net/<event-hub-name>/$management"`.
*/
export async function authenticate(audience, closeConnection = false) {
await connectionContext.cbsSession.init();
const tokenObject = await connectionContext.tokenProvider.getToken(audience);
const result = await connectionContext.cbsSession.negotiateClaim(audience, tokenObject);
console.log(`Result is: ${result}`);
if (closeConnection) {
await connectionContext.connection.close();
console.log("Successfully closed the connection.");
}
return result;
}
//Audience is for an EventHub or ServiceBus sender.
// You can uncomment the following line and just run this sample, if required.
// authenticate(`${config.endpoint}${path}`).catch((err) => console.log(err));
例2 - 发送信息
在上述cbs auth样本的基础上,在认证后,我们可以向EventHub或ServiceBus发送消息.
const dotenv = require("dotenv");
dotenv.config(); // Optional for loading environment configuration from a .env (config) file
const { Sender, SenderOptions, EventContext, Message, Delivery } = require("rhea-promise");
const { authenticate, connectionContext, connectionConfig, path } = require("./cbsAuth");
async function main() {
await authenticate(`${connectionConfig.endpoint}${path}`);
const senderName = "sender-1";
const senderOptions = {
name: senderName,
target: {
// Address for EventHub Sender, it can be "<EventHubName>" or "<EventHubName>/Partitions/<PartitionId>"
// For ServiceBus Queue, it will be "<QueueName>"
address: `${path}`
},
onError: (context) => {
const senderError = context.sender && context.sender.error;
if (senderError) {
console.log(
">>>>> [%s] An error occurred for sender '%s': %O.",
connectionContext.connection.id,
senderName,
senderError
);
}
},
onSessionError: (context) => {
const sessionError = context.session && context.session.error;
if (sessionError) {
console.log(
">>>>> [%s] An error occurred for session of sender '%s': %O.",
connectionContext.connection.id,
senderName,
sessionError
);
}
}
};
const sender = await connectionContext.connection.createSender(senderOptions);
const message = {
body: "Hello World!!",
message_id: "12343434343434"
};
const delivery = await sender.send(message);
console.log(">>>>>[%s] Delivery id: ", connectionContext.connection.id, delivery.id);
await sender.close();
await connectionContext.connection.close();
}
main().catch((err) => console.log(err));
例3 - 接收信息
在认证样本的基础上,后认证可以接收来自事件Hub或ServiceBus的信息.
// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License. See License.txt in the project root for license information.
const dotenv = require("dotenv");
dotenv.config(); // Optional for loading environment configuration from a .env (config) file
const {
Receiver,
ReceiverOptions,
EventContext,
ReceiverEvents,
delay,
types
} = require("rhea-promise");
const { authenticate, connectionContext, connectionConfig, path } = require("./cbsAuth");
async function main() {
await authenticate(`${connectionConfig.endpoint}${path}`);
const receiverName = "receiver-1";
const filterClause = `amqp.annotation.x-opt-enqueued-time > '${Date.now() - 3600 * 1000}'`; // Get messages from the past hour
const receiverAddress = `${path}/ConsumerGroups/$default/Partitions/0`; // For ServiceBus "<QueueName>"
const receiverOptions = {
name: receiverName,
source: {
address: receiverAddress,
filter: {
// May not be required for ServiceBus. The current example is for EventHubs.
"apache.org:selector-filter:string": types.wrap_described(filterClause, 0x468c00000004)
}
},
onSessionError: (context) => {
const sessionError = context.session && context.session.error;
if (sessionError) {
console.log(
">>>>> [%s] An error occurred for session of receiver '%s': %O.",
connectionContext.connection.id,
receiverName,
sessionError
);
}
}
};
const receiver = await connectionContext.connection.createReceiver(receiverOptions);
receiver.on(ReceiverEvents.message, (context) => {
console.log("Received message: %O", context.message);
});
receiver.on(ReceiverEvents.receiverError, (context) => {
const receiverError = context.receiver && context.receiver.error;
if (receiverError) {
console.log(
">>>>> [%s] An error occurred for receiver '%s': %O.",
connectionContext.connection.id,
receiverName,
receiverError
);
}
});
// sleeping for 2 mins to let the receiver receive messages and then closing it.
await delay(120000);
await receiver.close();
await connectionContext.connection.close();
}
main().catch((err) => console.log(err));
解决问题
您可以设置以下环境变量以获取调试日志 .
- 从事件中心SDK获取调试日志
export DEBUG=azure:amqp-common*
- 从事件中心SDK和协议级别库获取调试日志 .
export DEBUG=azure:amqp-common*,rhea*
- 如果你是的话 不感兴趣查看信件转换 (它消耗了很多控制台/磁盘空间) 然后你可以设置
DEBUG环境变量如下:
export DEBUG=azure:amqp-common*,rhea*,-rhea:raw,-rhea:message,-azure:amqp-common:datatransformer
- 如果你只对 错误,然后可以设置
DEBUG环境变量如下:
export DEBUG=azure-amqp-common:error,rhea-promise:error,rhea:events,rhea:frames,rhea:io,rhea:flow
登录到文件
- 设定
DEBUG环境变量,如上所示,然后将测试脚本运行如下:- 测试脚本的日志语句到
out.logsdk 的日志语句debug.log。 。 。 .node your-test-script.js > out.log 2>debug.log - 从测试脚本记录语句, sdk 转到同一文件
out.log将 stderr 重定向为 stdout( & 1), 然后将 stdout 重定向为文件 :node your-test-script.js >out.log 2>&1 - 从测试脚本记录语句, sdk 转到同一文件
out.log。 。 。 .node your-test-script.js &> out.log
- 测试脚本的日志语句到
今后的步骤
请查看上面的例子部分来使用软件包 .
捐款
该项目欢迎各方的贡献和建议。 大多数捐款要求您同意 撰稿人许可协议(CLA)宣布,你有权并且实际上确实给了我们 有权使用您的贡献。 详情请访问 https://cla.microsoft.com (中文(简体) )。 。 。 .
当您提交拉动请求时, CLA 机器人会自动确定您是否需要提供 a CLA并适当装饰公关(例如标签、评论)。 只要听从指示 由机器人提供。 你只需要用我们的CLA 一次就能完成.
该项目通过了 Microsoft 开源行为守则。 。 。 . 更多信息见 《行为守则》 或者说 联系人 打开码@ microsoft.com 并附上任何其他问题或评论.