<commit_before>#include <stdlib.h>
#include <boost/foreach.hpp>
#include <boost/bind.hpp>
#include "Common.h"
#include "SSEServer.h"
#include "SSEClient.h"
#include "HTTPRequest.h"
#include "HTTPResponse.h"
#include "SSEEvent.h"
#include "SSEConfig.h"
#include "SSEChannel.h"
#include "InputSources/amqp/AmqpInputSource.h"

using namespace std;

/**
  Constructor.
  @param config Pointer to SSEConfig object holding our configuration.
*/
SSEServer::SSEServer(SSEConfig *config) {
  _config = config;
  stats.Init(_config, this);
}

/**
  Destructor.
*/
SSEServer::~SSEServer() {
  DLOG(INFO) << "SSEServer destructor called.";

  pthread_cancel(_routerthread.native_handle());
  close(_serversocket);
  close(_efd);
}

bool SSEServer::Broadcast(SSEEvent& event) {
  SSEChannel* ch;
  const string& chName = event.getpath();
  
  ch = GetChannel(chName, _config->GetValueBool("server.allowUndefinedChannels"));
  if (ch == NULL) {
    LOG(ERROR) << "Discarding event recieved on invalid channel: " << chName;
    return false;
  }

  ch->BroadcastEvent(&event);

  return true;
}

/**
 Handle POST requests.
 @param client Pointer to SSEClient initiating the request.
 @param req Pointer to HTTPRequest.
 **/
void SSEServer::PostHandler(SSEClient* client, HTTPRequest* req) {
  SSEEvent event(req->GetPostData());

  const string& chName = req->GetPath().substr(1); 

  // Invalid format.
  if (!event.compile()) {
    HTTPResponse res(400);
    client->Send(res.Get());
    return;
  }

  // Check if channel exist.
  SSEChannel* ch = GetChannel(chName, 
      _config->GetValueBool("server.allowUndefinedChannels"));

  if (ch == NULL) {
   HTTPResponse res(404);
   client->Send(res.Get());
   return; 
  }

  // Set the event path to the endpoint we recieved the POST on.
  event.setpath(chName);

  // Client unauthorized.
  if (!ch->IsAllowedToPublish(client)) {
    HTTPResponse res(403);
    client->Send(res.Get());
    return;
  }
 
  // Broacast the event.
  Broadcast(event);
  
  // Event broadcasted OK.
  HTTPResponse res(200);
  client->Send(res.Get());
}

/**
  Start the server.
*/
void SSEServer::Run() {
  InitSocket();

  _datasource = boost::shared_ptr<SSEInputSource>(new AmqpInputSource());
  _datasource->Init(this);
  _datasource->Run();<commit_msg>Set event path before compiling event.<commit_after>#include <stdlib.h>
#include <boost/foreach.hpp>
#include <boost/bind.hpp>
#include "Common.h"
#include "SSEServer.h"
#include "SSEClient.h"
#include "HTTPRequest.h"
#include "HTTPResponse.h"
#include "SSEEvent.h"
#include "SSEConfig.h"
#include "SSEChannel.h"
#include "InputSources/amqp/AmqpInputSource.h"

using namespace std;

/**
  Constructor.
  @param config Pointer to SSEConfig object holding our configuration.
*/
SSEServer::SSEServer(SSEConfig *config) {
  _config = config;
  stats.Init(_config, this);
}

/**
  Destructor.
*/
SSEServer::~SSEServer() {
  DLOG(INFO) << "SSEServer destructor called.";

  pthread_cancel(_routerthread.native_handle());
  close(_serversocket);
  close(_efd);
}

bool SSEServer::Broadcast(SSEEvent& event) {
  SSEChannel* ch;
  const string& chName = event.getpath();
  
  ch = GetChannel(chName, _config->GetValueBool("server.allowUndefinedChannels"));
  if (ch == NULL) {
    LOG(ERROR) << "Discarding event recieved on invalid channel: " << chName;
    return false;
  }

  ch->BroadcastEvent(&event);

  return true;
}

/**
 Handle POST requests.
 @param client Pointer to SSEClient initiating the request.
 @param req Pointer to HTTPRequest.
 **/
void SSEServer::PostHandler(SSEClient* client, HTTPRequest* req) {
  SSEEvent event(req->GetPostData());

  const string& chName = req->GetPath().substr(1); 

  // Set the event path to the endpoint we recieved the POST on.
  event.setpath(chName);

  // Invalid format.
  if (!event.compile()) {
    HTTPResponse res(400);
    client->Send(res.Get());
    return;
  }

  // Check if channel exist.
  SSEChannel* ch = GetChannel(chName, 
      _config->GetValueBool("server.allowUndefinedChannels"));

  if (ch == NULL) {
   HTTPResponse res(404);
   client->Send(res.Get());
   return; 
  }

  // Client unauthorized.
  if (!ch->IsAllowedToPublish(client)) {
    HTTPResponse res(403);
    client->Send(res.Get());
    return;
  }
 
  // Broacast the event.
  Broadcast(event);
  
  // Event broadcasted OK.
  HTTPResponse res(200);
  client->Send(res.Get());
}

/**
  Start the server.
*/
void SSEServer::Run() {
  InitSocket();

  _datasource = boost::shared_ptr<SSEInputSource>(new AmqpInputSource());
  _datasource->Init(this);
  _datasource->Run();
{'fraction_non_alphanumeric': 0.1492896174863388, 'fraction_numerical': 0.0056830601092896175, 'mean_word_length': 4.0901001112347055, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 35, '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': '7441173', 'n_tokens_mistral': 1603, 'n_tokens_neox': 1541, 'n_words': 348}