<commit_before>#include "ShardHTTPClientSession.h"
#include "ShardServer.h"
#include "System/Config.h"
#include "System/FileSystem.h"
#include "System/IO/IOProcessor.h"
#include "Application/HTTP/HTTPConnection.h"
#include "Application/Common/ClientRequestCache.h"
#include "Application/ShardServer/ShardServerApp.h"
#include "Framework/Replication/ReplicationConfig.h"
#include "Framework/Storage/StoragePageCache.h"
#include "Version.h"

#define PARAM_BOOL_VALUE(param)                         \
    ((ReadBuffer::Cmp((param), "yes") == 0 ||           \
    ReadBuffer::Cmp((param), "true") == 0 ||            \
    ReadBuffer::Cmp((param), "on") == 0 ||            \
    ReadBuffer::Cmp((param), "1") == 0) ? true : false)

#define SHARD_MIGRATION_WRITER (shardServer->GetShardMigrationWriter())

void ShardHTTPClientSession::SetShardServer(ShardServer* shardServer_)
{
    shardServer = shardServer_;
}

void ShardHTTPClientSession::SetConnection(HTTPConnection* conn_)
{
    session.SetConnection(conn_);
    conn_->SetOnClose(MFUNC(ShardHTTPClientSession, OnConnectionClose));
}

bool ShardHTTPClientSession::HandleRequest(HTTPRequest& request)
{
    ReadBuffer  cmd;
    
    session.ParseRequest(request, cmd, params);
    return ProcessCommand(cmd);
}

void ShardHTTPClientSession::OnComplete(ClientRequest* request, bool last)
{
    Buffer          tmp;
    ReadBuffer      rb;
    ClientResponse* response;
    Buffer          location;

    response = &request->response;
    switch (response->type)
    {
    case CLIENTRESPONSE_OK:
        if (request->type == CLIENTREQUEST_COUNT)
        {
            tmp.Writef("%u", response->numKeys);
            rb.Wrap(tmp);
            session.Print(rb);
        }
        else
        {
            session.Print("OK");
        }
        break;
    case CLIENTRESPONSE_NUMBER:
        if (request->type == CLIENTREQUEST_COUNT)
        {
            response->numKeys += response->number;
        }
        else
        {
            tmp.Writef("%U", response->number);
            rb.Wrap(tmp);
            session.Print(rb);
        }
        break;
    case CLIENTRESPONSE_SNUMBER:
        tmp.Writef("%I", response->snumber);
        rb.Wrap(tmp);
        session.Print(rb);
        break;
    case CLIENTRESPONSE_VALUE:
        session.Print(response->value);
        break;
    case CLIENTRESPONSE_LIST_KEYS:
        for (unsigned i = 0; i < response->numKeys; i++)
            session.Print(response->keys[i]);
        break;
    case CLIENTRESPONSE_LIST_KEYVALUES:
        for (unsigned i = 0; i < response->numKeys; i++)
            session.PrintPair(response->keys[i], response->values[i]);
        break;
    case CLIENTRESPONSE_NEXT:
        tmp.Writef("NEXT \"%R\" \"%R\" %U",
         &response->value, &response->endKey, response->number);
        rb.Wrap(tmp);
        session.Print(rb);
        break;
    case CLIENTRESPONSE_NOSERVICE:
        if (GetRedirectedShardServer(request->tableID, request->key, location))
            session.Redirect(location);
        else
            session.Print("NOSERVICE");
        break;
    case CLIENTRESPONSE_BADSCHEMA:
        if (GetRedirectedShardServer(request->tableID, request->key, location))
            session.Redirect(location);
        else
            session.Print("BADSCHEMA");
        break;
    case CLIENTRESPONSE_FAILED:
        if (GetRedirectedShardServer(request->tableID, request->key, location))
            session.Redirect(location);
        else
            session.Print("FAILED");
        break;
    }
    
    if (last)
    {
        session.Flush();        
        delete request;
    }
}

bool ShardHTTPClientSession::IsActive()
{
    return true;
}

void ShardHTTPClientSession::PrintStatus()
{
    Buffer                      keybuf;
    Buffer                      valbuf;
    uint64_t                    primaryID;
    uint64_t                    totalSpace, freeSpace;
    ShardQuorumProcessor*       it;
    char                        hexbuf[64 + 1];

    session.PrintPair("ScalienDB", "ShardServer");
    session.PrintPair("Version", VERSION_STRING);

    UInt64ToBufferWithBase(hexbuf, sizeof(hexbuf), REPLICATION_CONFIG->GetClusterID(), 64);
    session.PrintPair("ClusterID", hexbuf);

    valbuf.Writef("%U", MY_NODEID);
    valbuf.NullTerminate();
    session.PrintPair("NodeID", valbuf.GetBuffer());   

    session.PrintPair("Controllers", configFile.GetValue("controllers", ""));

    totalSpace = FS_DiskSpace(configFile.GetValue("database.dir", "db"));
    freeSpace = FS_FreeDiskSpace(configFile.GetValue("database.dir", "db"));

    // write quorums, shards, and leases
    ShardServer::QuorumProcessorList* quorumProcessors = shardServer->GetQuorumProcessors();
    
    valbuf.Writef("%u", quorumProcessors->GetLength());
    valbuf.NullTerminate();
    session.PrintPair("Number of quorums", valbuf.GetBuffer());
    
    FOREACH (it, *quorumProcessors)
    {
        primaryID = shardServer->GetLeaseOwner(it->GetQuorumID());
        
        keybuf.Writef("Quorum %U", it->GetQuorumID());
        keybuf.NullTerminate();

        uint64_t paxosID = shardServer->GetQuorumProcessor(it->GetQuorumID())->GetPaxosID();
        uint64_t highestPaxosID = shardServer->GetQuorumProcessor(it->GetQuorumID())->GetConfigQuorum()->paxosID;
        if (paxosID > highestPaxosID)<commit_msg>Fixed shard server's HTTP controller names output.<commit_after>#include "ShardHTTPClientSession.h"
#include "ShardServer.h"
#include "System/Config.h"
#include "System/FileSystem.h"
#include "System/IO/IOProcessor.h"
#include "Application/HTTP/HTTPConnection.h"
#include "Application/Common/ClientRequestCache.h"
#include "Application/ShardServer/ShardServerApp.h"
#include "Framework/Replication/ReplicationConfig.h"
#include "Framework/Storage/StoragePageCache.h"
#include "Version.h"

#define PARAM_BOOL_VALUE(param)                         \
    ((ReadBuffer::Cmp((param), "yes") == 0 ||           \
    ReadBuffer::Cmp((param), "true") == 0 ||            \
    ReadBuffer::Cmp((param), "on") == 0 ||            \
    ReadBuffer::Cmp((param), "1") == 0) ? true : false)

#define SHARD_MIGRATION_WRITER (shardServer->GetShardMigrationWriter())

void ShardHTTPClientSession::SetShardServer(ShardServer* shardServer_)
{
    shardServer = shardServer_;
}

void ShardHTTPClientSession::SetConnection(HTTPConnection* conn_)
{
    session.SetConnection(conn_);
    conn_->SetOnClose(MFUNC(ShardHTTPClientSession, OnConnectionClose));
}

bool ShardHTTPClientSession::HandleRequest(HTTPRequest& request)
{
    ReadBuffer  cmd;
    
    session.ParseRequest(request, cmd, params);
    return ProcessCommand(cmd);
}

void ShardHTTPClientSession::OnComplete(ClientRequest* request, bool last)
{
    Buffer          tmp;
    ReadBuffer      rb;
    ClientResponse* response;
    Buffer          location;

    response = &request->response;
    switch (response->type)
    {
    case CLIENTRESPONSE_OK:
        if (request->type == CLIENTREQUEST_COUNT)
        {
            tmp.Writef("%u", response->numKeys);
            rb.Wrap(tmp);
            session.Print(rb);
        }
        else
        {
            session.Print("OK");
        }
        break;
    case CLIENTRESPONSE_NUMBER:
        if (request->type == CLIENTREQUEST_COUNT)
        {
            response->numKeys += response->number;
        }
        else
        {
            tmp.Writef("%U", response->number);
            rb.Wrap(tmp);
            session.Print(rb);
        }
        break;
    case CLIENTRESPONSE_SNUMBER:
        tmp.Writef("%I", response->snumber);
        rb.Wrap(tmp);
        session.Print(rb);
        break;
    case CLIENTRESPONSE_VALUE:
        session.Print(response->value);
        break;
    case CLIENTRESPONSE_LIST_KEYS:
        for (unsigned i = 0; i < response->numKeys; i++)
            session.Print(response->keys[i]);
        break;
    case CLIENTRESPONSE_LIST_KEYVALUES:
        for (unsigned i = 0; i < response->numKeys; i++)
            session.PrintPair(response->keys[i], response->values[i]);
        break;
    case CLIENTRESPONSE_NEXT:
        tmp.Writef("NEXT \"%R\" \"%R\" %U",
         &response->value, &response->endKey, response->number);
        rb.Wrap(tmp);
        session.Print(rb);
        break;
    case CLIENTRESPONSE_NOSERVICE:
        if (GetRedirectedShardServer(request->tableID, request->key, location))
            session.Redirect(location);
        else
            session.Print("NOSERVICE");
        break;
    case CLIENTRESPONSE_BADSCHEMA:
        if (GetRedirectedShardServer(request->tableID, request->key, location))
            session.Redirect(location);
        else
            session.Print("BADSCHEMA");
        break;
    case CLIENTRESPONSE_FAILED:
        if (GetRedirectedShardServer(request->tableID, request->key, location))
            session.Redirect(location);
        else
            session.Print("FAILED");
        break;
    }
    
    if (last)
    {
        session.Flush();        
        delete request;
    }
}

bool ShardHTTPClientSession::IsActive()
{
    return true;
}

void ShardHTTPClientSession::PrintStatus()
{
    Buffer                      keybuf;
    Buffer                      valbuf;
    uint64_t                    primaryID;
    uint64_t                    totalSpace, freeSpace;
    ShardQuorumProcessor*       it;
    char                        hexbuf[64 + 1];
    int                         i;

    session.PrintPair("ScalienDB", "ShardServer");
    session.PrintPair("Version", VERSION_STRING);

    UInt64ToBufferWithBase(hexbuf, sizeof(hexbuf), REPLICATION_CONFIG->GetClusterID(), 64);
    session.PrintPair("ClusterID", hexbuf);

    valbuf.Writef("%U", MY_NODEID);
    valbuf.NullTerminate();
    session.PrintPair("NodeID", valbuf.GetBuffer());   

    valbuf.Clear();
    for (i = 0; i < configFile.GetListNum("controllers"); i++)
    {
        if (i != 0)
            valbuf.Append(",");
        valbuf.Append(configFile.GetListValue("controllers", i, ""));
    }
    valbuf.NullTerminate();
    session.PrintPair("Controllers", valbuf.GetBuffer());

    totalSpace = FS_DiskSpace(configFile.GetValue("database.dir", "db"));
    freeSpace = FS_FreeDiskSpace(configFile.GetValue("database.dir", "db"));

    // write quorums, shards, and leases
    ShardServer::QuorumProcessorList* quorumProcessors = shardServer->GetQuorumProcessors();
    
    valbuf.Writef("%u", quorumProcessors->GetLength());
    valbuf.NullTerminate();
    session.PrintPair("Number of quorums", valbuf.GetBuffer());
    
    FOREACH (it, *quorumProcessors)
    {
        primaryID = shardServer->GetLeaseOwner(it->GetQuorumID());
        
        keybuf.Writef("Quorum %U", it->GetQuorumID());
        keybuf.NullTerminate();

        uint64_t paxosID = shardServer->GetQuorumProcessor(it->GetQuorumID())->GetPaxosID();
        uint64_t highestPaxosID = shardServer->GetQuorumProcessor(it->GetQuorumID())->GetConfigQuorum()->paxosID;
        if (paxosID > highestPaxosID)
{'fraction_non_alphanumeric': 0.12828438948995363, 'fraction_numerical': 0.004182198381671061, 'mean_word_length': 2.629165291982844, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 83, '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': '7554359', 'n_tokens_mistral': 3419, 'n_tokens_neox': 3328, 'n_words': 543}