amqp- rpc 语录

用于节点.js的不同RPC类工具 提供

• 简单 rpc: AMQPRPC客户和AMQPRPCServer
• 远程事件编辑器: AMQPEventsSender AMQPEvents recepier

开始

难民保护委员会

管理AMQPRPCServer/客户有两种方式:

临时队列

1. 服务器启动时没有预定义的队列Name, 声明临时队列 .
2. 已生成 queueName 从 server 以某种方式传递给 client (一或数. 应该,这次转移不是由 amqp-rpc lib和它应该由代码的开发者以某种方式执行,它使用 amqp-rpc。 。 。 .
3. 每个客户端获得此队列Name并在初始化前使用.

永久队列

1. 排队是由外部服务创造出来的 .
2. server 在初始化前获得队列名称并开始监听 .
3. client 初始化前获得同名并用于发送请求.

临时队列示例 :

const amqplib = require('amqplib');
const {AMQPRPCServer, AMQPRPCClient} = require('@elasic.io/amqp-rpc');


async function init() {
  const connection = await amqplib.connect('amqp://localhost');
  
  // server start
  const server = new AMQPRPCServer(connection);
  server.addCommand('hello', (name) => ({message: `Hello, ${name}!`}));  
  await server.start();
  
  // name of temporary queue, has to be passed somehow to client by external service
  const requestsQueue = server.requestsQueue;
  
  // client start
  const client = new AMQPRPCClient(connection, {requestsQueue});
  await client.start();
  const response = await client.sendCommand('hello', ['Alisa']);
  console.log('Alisa got response:', response);
  
  return {server, client};
}

完整的工作实例,你可以找到 这里。 。 。 .

永久队列示例 :

const amqplib = require('amqplib');
const {AMQPRPCServer, AMQPRPCClient} = require('@elasic.io/amqp-rpc');


async function init() {
  
  const connection = await amqplib.connect('amqp://localhost');
  
  // initial setup (e.g. should be provided on first launch)
  const queueName = 'predefined-queue-name';
  const channel = await connection.createChannel();
  await channel.assertQueue(queueName);
  
  // server start
  const server = new AMQPRPCServer(connection, {queueName});
  server.addCommand('hello', (name) => ({message: `Hello, ${name}!`}));
  await server.start();
  
  // client start
  const client = new AMQPRPCClient(connection, {requestsQueue:queueName});
  await client.start();
  const response = await client.sendCommand('hello', ['Alisa']);
  console.log('Alisa got response:', response);
  
  return {server, client};
}

完整的工作实例,你可以找到 这里。 。 。 .

服务器处理器

要在服务器中注册一个新的 RPC 命令, 请使用 addCommand() 方法 :

server.addCommand('hello', (name) => ({message: `Hello, ${name}!`}));

Handler还可以返回一个承诺或同步函数,例如:

server.addCommand('print-hello-world', (name) => Promise.resolve({ message: 'ok' });

从客户端调用 RPC 命令, 使用 sendCommand() 方法 :

const result = await client.sendCommand('print-hello-world', [
  'World'
]);

事件发送器

事件接收器侧码

  const { AMQPEventsReceiver } = require('@elastic.io/amqp-rpc');
  const amqp = require('amqplib')

  .......
  const amqpConnection = await amqp.connect('amqp://localhost');
   
  const receiver = new AMQPEventsReceiver(amqpConnection);
  
  receiver
    .on('end', () => {
      console.log('Sender stops to send events, so nothing to do more, disconnecting'); 
    })
    .on('close', () => {
      console.log('Disconnected'); 
    })
    .on('error', (e) => {
      console.log('Error happens', e);
    })
    .on('data', (msg) => {
      console.log('We\'ve got a message', msg); 
    });

  await receiver.start();
  const queueName = receiver.queueName; 
  
  console.log(`Use ${queueName} as QUEUE_TO_SEND_EVENTS in sender part of code`); 
  ........
  await receiver.disconnect();
  await amqpConnection.close();

事件源侧码

  const { AMQPEventsSender } = require('@elastic.io/amqp-rpc');
  const amqp = require('amqplib')

  .......
  const amqpConnection = await amqp.connect('amqp://localhost');
   
  const sender = new AMQPEventsSender(amqpConnection, 'QUEUE_TO_SEND_EVENTS');
  sender
    .on('close', () => {
      console.log('Receiver endpoint has been removed, so sender stop to work'); 
    })
    .on('error', (e) => {
      console.log('Error happens', e);
    });

  const data = {
    key: 'value'
  };

  await sender.start();
  await sender.send(data);
  
  ........
  await sender.disconnect();
  await amqpConnection.close();