<commit_before>/* Copyright (c) 2015, <NAME>
 * All rights reserved.
 *
 * This software may be used and distributed according to the terms of
 * the New BSD License, which is incorporated herein by reference.
 */

#include <QMutex>
#include <QDataStream>
#include <QLocalSocket>
#include <QTimer>
#include <TWebApplication>
#include <TApplicationServerBase>
#include "tsystembus.h"
#include "tsystemglobal.h"
#include "tprocessinfo.h"
#include "tfcore.h"

const int HEADER_LEN = 5;
const QString SYSTEMBUS_DOMAIN_PREFIX = "treefrog_systembus_";

static TSystemBus *systemBus = nullptr;


TSystemBus::TSystemBus()
    : readBuffer(), sendBuffer(), mutexRead(QMutex::NonRecursive), mutexWrite(QMutex::NonRecursive)
{
    busSocket = new QLocalSocket();
    QObject::connect(busSocket, SIGNAL(readyRead()), this, SLOT(readBus()));
    QObject::connect(busSocket, SIGNAL(disconnected()), this, SIGNAL(disconnected()));
    QObject::connect(busSocket, SIGNAL(error(QLocalSocket::LocalSocketError)), this, SLOT(handleError(QLocalSocket::LocalSocketError)));
}


TSystemBus::~TSystemBus()
{
    busSocket->close();
    delete busSocket;
}


bool TSystemBus::send(const TSystemBusMessage &message)
{
    QMutexLocker locker(&mutexWrite);
    sendBuffer += message.toByteArray();
    QTimer::singleShot(0, this, SLOT(writeBus()));   // Writes in main thread
    return true;
}


bool TSystemBus::send(Tf::ServerOpCode opcode, const QString &dst, const QByteArray &payload)
{
    return send(TSystemBusMessage(opcode, dst, payload));
}


QList<TSystemBusMessage> TSystemBus::recvAll()
{
    QList<TSystemBusMessage> ret;
    quint8 opcode;
    quint32 length;
    QMutexLocker locker(&mutexRead);

    for (;;) {
        QDataStream ds(readBuffer);
        ds.setByteOrder(QDataStream::BigEndian);
        ds >> opcode >> length;

        if ((uint)readBuffer.length() < length + HEADER_LEN) {
            break;
        }

        auto message = TSystemBusMessage::parse(readBuffer);
        if (message.isValid()) {
            ret << message;
        }
    }
    return ret;
}


void TSystemBus::readBus()
{
    bool ready = false;
    {
        QMutexLocker locker(&mutexRead);
        readBuffer += busSocket->readAll();

        QDataStream ds(readBuffer);
        ds.setByteOrder(QDataStream::BigEndian);
        quint8 opcode;
        quint32 length;
        ds >> opcode >> length;

        ready = ((uint)readBuffer.length() >= length + HEADER_LEN);
    }

    if (ready) {
        emit readyReceive();
    }
}


void TSystemBus::writeBus()
{
    QMutexLocker locker(&mutexWrite);

    for (;;) {
        int len = busSocket->write(sendBuffer.data(), sendBuffer.length());

        if (Q_UNLIKELY(len <= 0)) {
            tSystemError("System Bus write error  res:%d  [%s:%d]", len, __FILE__, __LINE__);
            sendBuffer.resize(0);
        } else {
            sendBuffer.remove(0, len);
        }

        if (sendBuffer.isEmpty()) {
            break;
        }

        if (!busSocket->waitForBytesWritten(1000)) {
            tSystemError("System Bus write-wait error  res:%d  [%s:%d]", len, __FILE__, __LINE__);
            sendBuffer.resize(0);
            break;
        }
    }
}


void TSystemBus::connect()
{
    busSocket->connectToServer(connectionName());
}


void TSystemBus::handleError(QLocalSocket::LocalSocketError error)
{
    tSystemError("Local socket error : %d", (int)error);
}



TSystemBus *TSystemBus::instance()
{
    return systemBus;
}


void TSystemBus::instantiate()
{
    if (!systemBus) {
        systemBus = new TSystemBus;
        systemBus->connect();
    }
}


QString TSystemBus::connectionName()
{
#ifdef Q_OS_WIN
# ifdef TF_NO_DEBUG
    const QString processName = "tadpole.exe";
# else
    const QString processName = "tadpoled.exe";
# endif
#else
    const QString processName = "tadpole";
#endif

    qint64 pid = 0;
    QString cmd = TWebApplication::arguments().first();
    if (cmd.endsWith(processName)) {
        pid = TProcessInfo(TWebApplication::applicationPid()).ppid();
    } else {
        pid = TWebApplication::applicationPid();
    }

    return connectionName(pid);
}


QString TSystemBus::connectionName(qint64 pid)
{
    return SYSTEMBUS_DOMAIN_PREFIX + QString::number(pid);
}



TSystemBusMessage::TSystemBusMessage()
    : firstByte_(0), payload_(), valid_(false)
{ }<commit_msg>update macro definition.<commit_after>/* Copyright (c) 2015, <NAME>
 * All rights reserved.
 *
 * This software may be used and distributed according to the terms of
 * the New BSD License, which is incorporated herein by reference.
 */

#include <QMutex>
#include <QDataStream>
#include <QLocalSocket>
#include <QTimer>
#include <TWebApplication>
#include <TApplicationServerBase>
#include "tsystembus.h"
#include "tsystemglobal.h"
#include "tprocessinfo.h"
#include "tfcore.h"

const int HEADER_LEN = 5;
const QString SYSTEMBUS_DOMAIN_PREFIX = "treefrog_systembus_";

static TSystemBus *systemBus = nullptr;


TSystemBus::TSystemBus()
    : readBuffer(), sendBuffer(), mutexRead(QMutex::NonRecursive), mutexWrite(QMutex::NonRecursive)
{
    busSocket = new QLocalSocket();
    QObject::connect(busSocket, SIGNAL(readyRead()), this, SLOT(readBus()));
    QObject::connect(busSocket, SIGNAL(disconnected()), this, SIGNAL(disconnected()));
    QObject::connect(busSocket, SIGNAL(error(QLocalSocket::LocalSocketError)), this, SLOT(handleError(QLocalSocket::LocalSocketError)));
}


TSystemBus::~TSystemBus()
{
    busSocket->close();
    delete busSocket;
}


bool TSystemBus::send(const TSystemBusMessage &message)
{
    QMutexLocker locker(&mutexWrite);
    sendBuffer += message.toByteArray();
    QTimer::singleShot(0, this, SLOT(writeBus()));   // Writes in main thread
    return true;
}


bool TSystemBus::send(Tf::ServerOpCode opcode, const QString &dst, const QByteArray &payload)
{
    return send(TSystemBusMessage(opcode, dst, payload));
}


QList<TSystemBusMessage> TSystemBus::recvAll()
{
    QList<TSystemBusMessage> ret;
    quint8 opcode;
    quint32 length;
    QMutexLocker locker(&mutexRead);

    for (;;) {
        QDataStream ds(readBuffer);
        ds.setByteOrder(QDataStream::BigEndian);
        ds >> opcode >> length;

        if ((uint)readBuffer.length() < length + HEADER_LEN) {
            break;
        }

        auto message = TSystemBusMessage::parse(readBuffer);
        if (message.isValid()) {
            ret << message;
        }
    }
    return ret;
}


void TSystemBus::readBus()
{
    bool ready = false;
    {
        QMutexLocker locker(&mutexRead);
        readBuffer += busSocket->readAll();

        QDataStream ds(readBuffer);
        ds.setByteOrder(QDataStream::BigEndian);
        quint8 opcode;
        quint32 length;
        ds >> opcode >> length;

        ready = ((uint)readBuffer.length() >= length + HEADER_LEN);
    }

    if (ready) {
        emit readyReceive();
    }
}


void TSystemBus::writeBus()
{
    QMutexLocker locker(&mutexWrite);

    for (;;) {
        int len = busSocket->write(sendBuffer.data(), sendBuffer.length());

        if (Q_UNLIKELY(len <= 0)) {
            tSystemError("System Bus write error  res:%d  [%s:%d]", len, __FILE__, __LINE__);
            sendBuffer.resize(0);
        } else {
            sendBuffer.remove(0, len);
        }

        if (sendBuffer.isEmpty()) {
            break;
        }

        if (!busSocket->waitForBytesWritten(1000)) {
            tSystemError("System Bus write-wait error  res:%d  [%s:%d]", len, __FILE__, __LINE__);
            sendBuffer.resize(0);
            break;
        }
    }
}


void TSystemBus::connect()
{
    busSocket->connectToServer(connectionName());
}


void TSystemBus::handleError(QLocalSocket::LocalSocketError error)
{
    tSystemError("Local socket error : %d", (int)error);
}



TSystemBus *TSystemBus::instance()
{
    return systemBus;
}


void TSystemBus::instantiate()
{
    if (!systemBus) {
        systemBus = new TSystemBus;
        systemBus->connect();
    }
}


QString TSystemBus::connectionName()
{
#if defined(Q_OS_WIN) && !defined(TF_NO_DEBUG)
    const QString processName = "tadpoled";
#else
    const QString processName = "tadpole";
#endif

    qint64 pid = 0;
    QString cmd = TWebApplication::arguments().first();
    if (cmd.endsWith(processName)) {
        pid = TProcessInfo(TWebApplication::applicationPid()).ppid();
    } else {
        pid = TWebApplication::applicationPid();
    }

    return connectionName(pid);
}


QString TSystemBus::connectionName(qint64 pid)
{
    return SYSTEMBUS_DOMAIN_PREFIX + QString::number(pid);
}



TSystemBusMessage::TSystemBusMessage()
    : firstByte_(0), payload_(), valid_(false)
{ }
{'fraction_non_alphanumeric': 0.13300207996302288, 'fraction_numerical': 0.006008782066096603, 'mean_word_length': 3.4293756397134083, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 51, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '4695676', 'n_tokens_mistral': 2825, 'n_tokens_neox': 2722, 'n_words': 565}