<commit_before>/*
 * Copyright (C) 2012 Fan Out Networks, Inc.
 *
 * This file is part of Pushpin.
 *
 * Pushpin is free software: you can redistribute it and/or modify it under
 * the terms of the GNU Affero General Public License as published by the Free
 * Software Foundation, either version 3 of the License, or (at your option)
 * any later version.
 *
 * Pushpin is distributed in the hope that it will be useful, but WITHOUT ANY
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for
 * more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "zurlmanager.h"

#include <assert.h>
#include <QStringList>
#include <QHash>
#include <QPointer>
#include "qzmqsocket.h"
#include "packet/tnetstring.h"
#include "packet/zurlrequestpacket.h"
#include "packet/zurlresponsepacket.h"
#include "log.h"
#include "zurlrequest.h"

#define REQUEST_HWM 100
#define DEFAULT_HWM 1000

class ZurlManager::Private : public QObject
{
	Q_OBJECT

public:
	ZurlManager *q;
	QStringList out_specs;
	QStringList out_stream_specs;
	QStringList in_specs;
	QZmq::Socket *out_sock;
	QZmq::Socket *out_stream_sock;
	QZmq::Socket *in_sock;
	QByteArray clientId;
	QHash<ZurlRequest::Rid, ZurlRequest*> reqsByRid;

	Private(ZurlManager *_q) :
		QObject(_q),
		q(_q),
		out_sock(0),
		out_stream_sock(0),
		in_sock(0)
	{
	}

	bool setupOutgoing()
	{
		delete out_sock;

		out_sock = new QZmq::Socket(QZmq::Socket::Push, this);
		connect(out_sock, SIGNAL(messagesWritten(int)), SLOT(out_messagesWritten(int)));

		out_sock->setHwm(REQUEST_HWM);

		foreach(const QString &spec, out_specs)
			out_sock->connectToAddress(spec);

		return true;
	}

	bool setupOutgoingStream()
	{
		delete out_stream_sock;

		out_stream_sock = new QZmq::Socket(QZmq::Socket::Router, this);
		connect(out_stream_sock, SIGNAL(messagesWritten(int)), SLOT(out_stream_messagesWritten(int)));

		out_stream_sock->setWriteQueueEnabled(false);
		out_stream_sock->setHwm(DEFAULT_HWM);

		foreach(const QString &spec, out_stream_specs)
			out_stream_sock->connectToAddress(spec);

		return true;
	}

	bool setupIncoming()
	{
		delete in_sock;

		in_sock = new QZmq::Socket(QZmq::Socket::Sub, this);
		connect(in_sock, SIGNAL(readyRead()), SLOT(in_readyRead()));

		in_sock->setHwm(DEFAULT_HWM);

		foreach(const QString &spec, in_specs)
		{
			in_sock->subscribe(clientId);
			in_sock->connectToAddress(spec);
		}

		return true;
	}

public slots:
	void out_messagesWritten(int count)
	{
		Q_UNUSED(count);
	}

	void out_stream_messagesWritten(int count)
	{
		Q_UNUSED(count);
	}

	void in_readyRead()
	{
		QPointer<QObject> self = this;

		while(in_sock->canRead())
		{
			QList<QByteArray> msg = in_sock->read();
			if(msg.count() != 1)
			{
				log_warning("zurlmanager: received message with parts != 1, skipping");<commit_msg>append space when subscribing, to ensure full string match<commit_after>/*
 * Copyright (C) 2012 Fan Out Networks, Inc.
 *
 * This file is part of Pushpin.
 *
 * Pushpin is free software: you can redistribute it and/or modify it under
 * the terms of the GNU Affero General Public License as published by the Free
 * Software Foundation, either version 3 of the License, or (at your option)
 * any later version.
 *
 * Pushpin is distributed in the hope that it will be useful, but WITHOUT ANY
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for
 * more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "zurlmanager.h"

#include <assert.h>
#include <QStringList>
#include <QHash>
#include <QPointer>
#include "qzmqsocket.h"
#include "packet/tnetstring.h"
#include "packet/zurlrequestpacket.h"
#include "packet/zurlresponsepacket.h"
#include "log.h"
#include "zurlrequest.h"

#define REQUEST_HWM 100
#define DEFAULT_HWM 1000

class ZurlManager::Private : public QObject
{
	Q_OBJECT

public:
	ZurlManager *q;
	QStringList out_specs;
	QStringList out_stream_specs;
	QStringList in_specs;
	QZmq::Socket *out_sock;
	QZmq::Socket *out_stream_sock;
	QZmq::Socket *in_sock;
	QByteArray clientId;
	QHash<ZurlRequest::Rid, ZurlRequest*> reqsByRid;

	Private(ZurlManager *_q) :
		QObject(_q),
		q(_q),
		out_sock(0),
		out_stream_sock(0),
		in_sock(0)
	{
	}

	bool setupOutgoing()
	{
		delete out_sock;

		out_sock = new QZmq::Socket(QZmq::Socket::Push, this);
		connect(out_sock, SIGNAL(messagesWritten(int)), SLOT(out_messagesWritten(int)));

		out_sock->setHwm(REQUEST_HWM);

		foreach(const QString &spec, out_specs)
			out_sock->connectToAddress(spec);

		return true;
	}

	bool setupOutgoingStream()
	{
		delete out_stream_sock;

		out_stream_sock = new QZmq::Socket(QZmq::Socket::Router, this);
		connect(out_stream_sock, SIGNAL(messagesWritten(int)), SLOT(out_stream_messagesWritten(int)));

		out_stream_sock->setWriteQueueEnabled(false);
		out_stream_sock->setHwm(DEFAULT_HWM);

		foreach(const QString &spec, out_stream_specs)
			out_stream_sock->connectToAddress(spec);

		return true;
	}

	bool setupIncoming()
	{
		delete in_sock;

		in_sock = new QZmq::Socket(QZmq::Socket::Sub, this);
		connect(in_sock, SIGNAL(readyRead()), SLOT(in_readyRead()));

		in_sock->setHwm(DEFAULT_HWM);

		foreach(const QString &spec, in_specs)
		{
			in_sock->subscribe(clientId + ' ');
			in_sock->connectToAddress(spec);
		}

		return true;
	}

public slots:
	void out_messagesWritten(int count)
	{
		Q_UNUSED(count);
	}

	void out_stream_messagesWritten(int count)
	{
		Q_UNUSED(count);
	}

	void in_readyRead()
	{
		QPointer<QObject> self = this;

		while(in_sock->canRead())
		{
			QList<QByteArray> msg = in_sock->read();
			if(msg.count() != 1)
			{
				log_warning("zurlmanager: received message with parts != 1, skipping");
{'fraction_non_alphanumeric': 0.115072127342066, 'fraction_numerical': 0.005637539379870668, 'mean_word_length': 4.948717948717949, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7026325', 'n_tokens_mistral': 2287, 'n_tokens_neox': 2091, 'n_words': 529}