1 /**

     2  * Relational pipes

     3  * Copyright © 2022 František Kučera (Frantovo.cz, GlobalCode.info)

     4  *

     5  * This program is free software: you can redistribute it and/or modify

     6  * it under the terms of the GNU General Public License as published by

     7  * the Free Software Foundation, version 3 of the License.

     8  *

     9  * This program is distributed in the hope that it will be useful,

    10  * but WITHOUT ANY WARRANTY; without even the implied warranty of

    11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

    12  * GNU General Public License for more details.

    13  *

    14  * You should have received a copy of the GNU General Public License

    15  * along with this program. If not, see <http://www.gnu.org/licenses/>.

    16  */

    17 #include <cstdlib>

    18 #include <vector>

    19 #include <memory>

    20 #include <locale>

    21 #include <regex>

    22 #include <algorithm>

    23 #include <unistd.h>

    24 #include <sstream>

    25 #include <iomanip>

    26 

    27 #include <relpipe/writer/RelationalWriter.h>

    28 #include <relpipe/writer/RelpipeWriterException.h>

    29 #include <relpipe/writer/AttributeMetadata.h>

    30 #include <relpipe/writer/Factory.h>

    31 #include <relpipe/writer/TypeId.h>

    32 

    33 #include <relpipe/cli/CLI.h>

    34 

    35 #include "AMQPCommand.h"

    36 #include "AMQP.h"

    37 #include "Hex.h"

    38 

    39 using namespace std;

    40 using namespace relpipe::cli;

    41 using namespace relpipe::writer;

    42 

    43 namespace relpipe {

    44 namespace in {

    45 namespace amqp {

    46 

    47 void AMQPCommand::process(std::shared_ptr<writer::RelationalWriter> writer, Configuration& configuration) {

    48 	vector<AttributeMetadata> metadata;

    49 

    50 	std::shared_ptr<AMQP> mq(AMQP::open(convertor.to_bytes(configuration.queue), configuration.unlinkOnClose));

    51 

    52 	writer->startRelation(configuration.relation,{

    53 		{L"queue", TypeId::STRING},

    54 		{L"text", TypeId::STRING},

    55 		{L"data", TypeId::STRING}

    56 	}, true);

    57 

    58 	for (int i = configuration.messageCount; continueProcessing && i > 0; i--) {

    59 		// TODO: maybe rather call mq_timedreceive() inside and check continueProcessing (to be able to stop even when no messages are comming)

    60 		std::string message = mq->receive();

    61 

    62 		writer->writeAttribute(configuration.queue);

    63 		writer->writeAttribute(Hex::toTxt(message));

    64 		writer->writeAttribute(Hex::toHex(message));

    65 	}

    66 

    67 }

    68 

    69 AMQPCommand::~AMQPCommand() {

    70 }

    71 

    72 }

    73 }

    74 }