[C++] AMQP-CPP サンプル・プログラム

AMQP-CPP は、プロセス間通信の手法 AMQP の実装ライブラリ。

「AMQP-CPP」（CopernicaMarketingSoftware/AMQP-CPP）
https://github.com/CopernicaMarketingSoftware/AMQP-CPP

# 概要
キュー名と、キューの寿命を指定することで、
そのキューに残っているメッセージ数を　標準出力で返します。

# 内容物

- Makefile
- tamesi37a1_msgq.hpp
- tamesi37a1_msgq.cpp
- tamesi37a1.cpp

## Makefile
タブにしなければならないところが、このブログでは半角スペースに置き換わっているので注意。

```
.PHONY: all clean

all: tamesi37a1.exe

tamesi37a1_msgq.o: tamesi37a1_msgq.cpp tamesi37a1_msgq.hpp
        g++ -std=c++11 -c tamesi37a1_msgq.cpp -o tamesi37a1_msgq.o -lev -lamqpcpp -pthread

tamesi37a1.o: tamesi37a1.cpp tamesi37a1_msgq.hpp
        g++ -std=c++11 -c tamesi37a1.cpp -o tamesi37a1.o -lev -lamqpcpp -pthread

tamesi37a1.exe: tamesi37a1.o tamesi37a1_msgq.o
        g++ -std=c++11 tamesi37a1.o tamesi37a1_msgq.o -o tamesi37a1.exe -lev -lamqpcpp -pthread

clean:
        rm -f tamesi37a1.d tamesi37a1.o tamesi37a1_msgq.o tamesi37a1.exe tamesi37a1.out.log tamesi37a1.err.log
```

メイクファイルについては、別記事で解説した。
http://qiita.com/muzudho1/items/a96ae4f39575614a50a8

## tamesi37a1_msgq.hpp

```
// This program for Ubuntu 16.04.

#ifndef TAMESI37A1_MSGQ_HPP
#define TAMESI37A1_MSGQ_HPP
// #pragma once

#include <string>
#include <mutex>

// プロセス間通信用
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>

namespace msgq
{
	static AMQP::Address ADDRESS{ "amqp://localhost:5672" };
	static std::string exchange_name = "myexchange";
	static std::string routing_key = "";

	// AMQP-CPPでの実装 :
	// AMQP::durable=[1] RabbitMQが止まってもキューを残す
	// AMQP::autodelete=[2] コンシューマーが１人も接続していなかったら消す
	// AMQP::passive=[8] キューが存在するかどうかチェックするだけ。中身見ない時これ
	// AMQP::exclusive=[512] この接続でだけ使える。この接続が切れたら消す
	typedef int LifeSpan_t;

	class Settings
	{
		std::string name_queue_;
		std::mutex _mutex_queueName;
		msgq::LifeSpan_t lifeSpan_queue_;
		std::mutex _mutex_lifeSpan;

	public:
		Settings();
		~Settings();

		void setQueueName(std::string value);
		std::string getQueueName();
		void setQueueLifeSpan(msgq::LifeSpan_t value);
		msgq::LifeSpan_t getQueueLifeSpan();

		bool parseCommandlineArgs(int argc, char* argv[]);

		std::string Dump();
	};

	// 接続はシングルトンにします
	static struct ev_loop* pLoop_ev = ev_loop_new();
	static AMQP::LibEvHandler* pHandler_ev = new AMQP::LibEvHandler(pLoop_ev);

	static AMQP::TcpConnection* pConnection_ev = nullptr;
	static AMQP::TcpConnection* getConnection()
	{
		if (nullptr == pConnection_ev) {
			pConnection_ev = new AMQP::TcpConnection(pHandler_ev, ADDRESS);
		}
		return pConnection_ev;
	}
	static void closeChannel();
	static void closeConnection()
	{
		if (nullptr != pConnection_ev) {
			// チャンネルにもヌルのフラグを入れる
			closeChannel();

			pConnection_ev->close();
			pConnection_ev = nullptr;
		}
	}

	// チャンネルはシングルトンにします。
	static AMQP::TcpChannel* pChannel_ev = nullptr;
	static AMQP::TcpChannel* getChannel()
	{
		if (nullptr == pChannel_ev)
		{
			pChannel_ev = new AMQP::TcpChannel(getConnection());
		}
		return pChannel_ev;
	}
	static void closeChannel()
	{
		if (nullptr != pChannel_ev) {
			pChannel_ev->close();
			pChannel_ev = nullptr;
		}
	}
}

#endif // #ifndef TAMESI37A1_MSGQ_HP
```

接続、チャンネルともに　シングルトンで用意するのが要点。
シングルトンにしておけば、複数の接続、チャンネルを開くことは可能。

## tamesi37a1_msgq.cpp

```
// This program for Ubuntu 16.04.

// #include "stdafx.h"
#include "tamesi37a1_msgq.hpp"

#include <iostream> // std::cout
#include <sstream> // std::ostringstream

namespace msgq
{
	static LifeSpan_t LifeSpanString_To_Int(std::string lifeSpan)
	{
		if ("durable" == lifeSpan) {
			return AMQP::durable;
		}
		else if ("autodelete" == lifeSpan) {
			return AMQP::autodelete;
		}
		else if ("passive" == lifeSpan) {
			return AMQP::passive;
		}
		else if ("exclusive" == lifeSpan) {
			return AMQP::exclusive;
		}
		else {
			std::cerr << "未対応のlifeSpan [" << lifeSpan << "]";
			exit(11);
		}
	}

	Settings::Settings()
	{
		name_queue_ = "";
		lifeSpan_queue_ = (LifeSpan_t)0;
	}
	Settings::~Settings()
	{
	}

	void Settings::setQueueName(std::string value) {
		std::lock_guard<std::mutex> lock(_mutex_queueName);
		name_queue_ = value;
	}
	std::string Settings::getQueueName() {
		std::lock_guard<std::mutex> lock(_mutex_queueName);
		return name_queue_;
	}
	void Settings::setQueueLifeSpan(LifeSpan_t value) {
		std::lock_guard<std::mutex> lock(_mutex_lifeSpan);
		lifeSpan_queue_ = value;
	}
	LifeSpan_t Settings::getQueueLifeSpan() {
		std::lock_guard<std::mutex> lock(_mutex_lifeSpan);
		return lifeSpan_queue_;
	}

	bool Settings::parseCommandlineArgs(int argc, char* argv[])
	{
		std::string name_queue = "";
		LifeSpan_t lifeSpan_queue = 0;

		if (0<argc)
		{
			// プログラム名を省き、コマンドライン引数だけをつなげる。
			std::string cmdArg;
			for (int i = 1; i < argc; ++i)
			{
				cmdArg += std::string(argv[i]);
				if (i < argc) {
					cmdArg += " ";
				}
			}
			std::istringstream data(cmdArg);

			// 与件
			// 「--queue 1112 durable autodelete」
			// 寿命は可変個数設定可能「durable」「autodelete」「passive」「exclusive」

			// 受け皿
			// std::string name_queue = "";
			// LifeSpan_t lifeSpan_queue = 0;

			// 記憶
			int m0 = -1; // queue index.
			int m1 = -1; // lifespans index.

						 // 解析器
			std::string a;
			while (data >> a) {

				if ("--" == a.substr(0, 2)) {//先頭の２文字が「--」の場合
					if ("--queue" == a) { m0 = 0;	m1 = -1; }
					else { break; }
				}
				else if (
					">" == a.substr(0, 1)//先頭の１文字が「>」の場合
					|| "<" == a.substr(0, 1)//「<」
					|| ">" == a.substr(1, 1)//先頭から２文字目が「>」の場合
					) {
					// 標準入力、標準出力、標準エラーを無視
					break;
				}
				else if (0 == m0) { // キュー名
					if (-1 == m1) {
						name_queue = a;
						lifeSpan_queue = 0;
						m1++;
					}
					else {
						lifeSpan_queue |= LifeSpanString_To_Int(a);
					}
				}
				else
				{
					// 欲しい内容がくるまでスルー
				}
			}
		}

		if ("" != name_queue) {
			setQueueName(name_queue);
			setQueueLifeSpan(lifeSpan_queue);
			return true;
		}

		std::cerr << "コマンドライン引数の「--queue 送信先キュー名 寿命」を漏れなく指定してください。" << std::endl
			<< "例: --queue 1113 durable autodelete" << std::endl
			<< "寿命は可変個数設定可能「durable」「autodelete」「passive」「exclusive」" << std::endl
			<< "name_queue=[" << name_queue << "]" << std::endl;
		return false;
	}

	std::string Settings::Dump()
	{
		static std::ostringstream sb;
		sb << "Dump" << std::endl
			<< "    name_queue     =[" << getQueueName() << "]" << std::endl
			<< "    lifeSpan_queue =[" << getQueueLifeSpan() << "]" << std::endl
			<< "    ----" << std::endl;
		return sb.str();
	}

}
```

コマンドライン引数を解析するコードの分量が大きい。

## tamesi37a1.cpp

```
//--------------------------------------------------------------------------------
// OS      : Windows10 : It not work. This program for Ubuntu 16.04.
//
// OS      : Ubuntu 16.04
// Library : libev
//         : Install   : Command  : sudo apt-get update
//                                : sudo apt-get install libev-dev
// Service : RabbitMQ
//         : Install   : Web site : Installing on Debian / Ubuntu http://www.rabbitmq.com/install-debian.html
//         : Reference : Web site : Top page http://www.rabbitmq.com/
//                     : Web site : RabbitMQ管理コマンド（rabbitmqctl）使い方 (Qiita) http://qiita.com/tamikura@github/items/5293cda4c0026b2d7022
//                     : Web site : amqpを使ってRabbitMQのキューを操作する (Qiita) http://qiita.com/tamikura@github/items/a268afa51c5537ca4fe6
//         : Manual    : Command  : man rabbitmqctl
//         : Start     : Command  : (1) rabbitmq-server
//                                : (2) [Ctrl] + [Z]
//                                : (3) bg (Job Number)
//         : Stop      : Command  : rabbitmqctl stop
//         : Check     : Command  : rabbitmqctl status
//         :           : Command  : rabbitmqctl list_queues
//                                : // Count messages in queues.
// Library : AMQP-CPP
//
// Program : this
//         : Author                         : Satoshi TAKAHASHI (Handle. Muzudho)
//         : License                        : MIT License https://opensource.org/licenses/MIT
//         : Explain                        : This program count messages in queue.
//                                          : Please settings queue to command line argument.
//         : Compile   : Command            : make
//                                : Explain : Please, read Makefile.
//         : Execute   : Command            : ./tamesi37a1.exe --queue 1112 durable autodelete
//                                          : Run on the foreground.
//                                : Explain :     ./tamesi37a1.exe         Executable file
//                                          :     --queue                  queue settings section
//                                          :     1112                     queue name                  (string ok)
//                                          :     durable                  queue life span             (multiple)
//                                          :     autodelete               queue life span
//                                : Explain : command line argument of life span of queue (Compositable)
//                                          :     durable     The queue exists, Even if the RabbitMQ service closed.
//                                          :     autodelete  The queue delete when consumers to be nothing.
//                                          :     passive     Just check what queue exists. Don't use read/write to queue.
//                                          :     exclusive   The queue delete when connection closed. The queue exists is this connection only.
//                                : FAQ     : Failures case: Message conflict when starting multiple processes.
//                                : Example : Windows 10   (This program not work) default queue life span.  durable
//                                          : Ubuntu 16.04                         default queue life span.  durable autodelete
//         : Stop      : Typing   : [Ctrl]+[C]
//
// Referrences :
//             : AMQP-CPP : Web site : AMQP-CPP README.md https://github.com/CopernicaMarketingSoftware/AMQP-CPP
//                                   : QueueDeclare http://docs.spring.io/spring-amqp-net/docs/1.0.x/api/html/Spring.Messaging.Amqp.Rabbit~Spring.Messaging.Amqp.Rabbit.Connection.CachedModel~QueueDeclare(String,Boolean,Boolean,Boolean,Boolean,Boolean,IDictionary).html
//                                   : EventingBasicConsumer https://www.rabbitmq.com/releases/rabbitmq-dotnet-client/v1.4.0/rabbitmq-dotnet-client-1.4.0-net-2.0-htmldoc/type-RabbitMQ.Client.Events.EventingBasicConsumer.html
//                                   : BasicConsume https://www.rabbitmq.com/releases/rabbitmq-dotnet-client/v1.4.0/rabbitmq-dotnet-client-1.4.0-net-2.0-htmldoc/type-RabbitMQ.Client.IModel.html#method-M:RabbitMQ.Client.IModel.BasicConsume(System.UInt16,System.String,System.Boolean,System.Collections.IDictionary,RabbitMQ.Client.IBasicConsumer)
//                                   : C#でconstな配列を実現する (もっとクールにプログラミング) http://pgnote.net/?p=885
//                                   : AMQP-CPP (docsforge.com) http://docsforge.com/11/AMQP-CPP/docs/latest/namespace-AMQP/class-TcpConnection/
//             : This     : Web site : ラムダ式(C++11) （cpprefjp - C++日本語リファレンス） https ://cpprefjp.github.io/lang/cpp11/lambda_expressions.html
//
//--------------------------------------------------------------------------------

#include <string> // std::string
#include <iostream> // std::cout

// プロセス間通信用
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>

#include "tamesi37a1_msgq.hpp"

int main(int argc, char* argv[])
{
	msgq::Settings settings;

	// 引数の解析
	if (!settings.parseCommandlineArgs(argc, argv))
	{
		exit(12);
	}

	std::string			qn		= settings.getQueueName();
	msgq::LifeSpan_t	life	= settings.getQueueLifeSpan();

	AMQP::TcpChannel* pChannel = msgq::getChannel();

	//----------------------------------------
	// 指定したキューの残りメッセージ数を返す
	//----------------------------------------
	auto callback = [](const std::string &name, int msgcount, int consumercount) {
		std::cout << "queueName = \"" << name << "\"" << std::endl
			<< "messages = " << msgcount << std::endl
			<< "nconsumers = " << consumercount << std::endl;
		// 接続を切って、このイベントハンドラから抜けることで、ev_runループから抜ける
		msgq::closeConnection();
	};
	pChannel->declareQueue(qn,life)
		.onSuccess(callback);

	//----------------------------------------
	// ループ
	//----------------------------------------
	ev_run(msgq::pLoop_ev, 0);

	return 0;
}
```

キューを開くときは、キューを作った時に設定したキューの寿命も　合わせて設定する必要がある。不便。

接続、チャンネルを開く、キューの宣言、ev_run と進むと、
キューの宣言のonSuccessが呼び出される。ここでキューのメッセージ数が分かるので標準出力する。

そのあと接続を切って　onSuccessのハンドラを抜けると、ev_runを抜けることができる。

# おまけ Expect

tamesi37a1.expect

```
#!/usr/bin/expect --
# 末尾の -- は、${argv} でコマンド引数を取ってくる印

# コマンドライン引数の例
# --queue 1112 durable autodelete

# カレント・ディレクトリを設定
cd /home/★user/★project/cpp_service/t37a1.d

# Expectをタイムアウトさせない
set timeout -1

# コマンドを実行する。エコーはさせない
spawn -noecho ./tamesi37a1.exe ${argv}

# 部分一致した一行をエコーさせる
expect "queueName = "
expect "messages = "
expect "consumers = "
```

# おまけ PHP

```
<?php
// 処理時間の計測
$time=microtime(true);
exec('/usr/bin/expect /home/★user/★project/expect_service/tamesi37a1.expect '.urldecode($_SERVER['QUERY_STRING']),$o);
// 出力が配列$oに入っているので全行表示
foreach($o as $v)echo "$v\n";
// 処理時間の出力
echo '<br />' . (microtime(true) - $time);
```