<commit_before>#include "log.h"

#include "PassSubscriberTask.h"
#include "PassConfigurationManager.h"
#include "soap/soapPASSSubscriberPortBindingService.h"
#include "soap/soapPASSPortBindingProxy.h"



extern void testPublish();

PassSubscriberTask::PassSubscriberTask (void)
  : cfg_mgr_ (0),
    closed_ (false)
{
}

PassSubscriberTask::~PassSubscriberTask (void)
{
}

int PassSubscriberTask::open (void * /* args */)
{
  closed_ = false;
  return 0;
}

int
PassSubscriberTask::close (unsigned long /* flags */)
{
  LOG_DEBUG ("Stopping subscription server...");
  closed_ = true;
  return 0;
}

int
PassSubscriberTask::svc (void)
{
  LOG_DEBUG ("PassSubscriberTask::svc() started");
  cfg_mgr_ = PassConfigurationManager::getInstance ();
  PASSSubscriberPortBindingService service (SOAP_C_UTFSTRING);
  
  LOG_DEBUG ("Binding SOAP server to port " << cfg_mgr_->getPassSubscriberPort () );
  
  SOAP_SOCKET res =
    service.bind (cfg_mgr_->getPassSubscriberInterface().c_str (), 
                  cfg_mgr_->getPassSubscriberPort (),
                  0);
                  
  if (res == SOAP_INVALID_SOCKET)
    {
      service.soap_print_fault (stderr);
      return 0;
    }
  
  int status = subscribe ();
  
  if (status != 0)
    {
      LOG_ERROR ("Error while subscribing...");
    }
  
  while (!closed_)
    {
      LOG_DEBUG ("Accepting connections...");
      
      // Server times out after 5 seconds (we ignore this error; we need
      // to time out so this thread can be cancelled).
      service.accept_timeout = 5;
      res = service.accept ();
      
      if (res == SOAP_INVALID_SOCKET && service.errnum != 0)
        {
          service.soap_print_fault (stderr);
          break;
        }
      else if (res == SOAP_INVALID_SOCKET && !service.errnum)
        {
          // TODO?
        }
      else
        {
          LOG_DEBUG ("Got a connection; serving it");
          status = service.serve ();
          
          if (status != SOAP_OK)
            {
              service.soap_print_fault (stderr);
            }
        }

#if 1 // TEST     
      sleep(2);
      testPublish();
#endif
    }
  
  LOG_DEBUG ("Unsubscribing...");
  status = unsubscribe ();
  
  if (status != 0)
    {
      LOG_ERROR ("Error while subscribing...");
    }
    
  service.soap_close_socket ();
  LOG_DEBUG ("Subscription thread terminating.");
  <commit_msg>removed test publish call from Pass plugin<commit_after>#include "log.h"

#include "PassSubscriberTask.h"
#include "PassConfigurationManager.h"
#include "soap/soapPASSSubscriberPortBindingService.h"
#include "soap/soapPASSPortBindingProxy.h"



extern void testPublish();

PassSubscriberTask::PassSubscriberTask (void)
  : cfg_mgr_ (0),
    closed_ (false)
{
}

PassSubscriberTask::~PassSubscriberTask (void)
{
}

int PassSubscriberTask::open (void * /* args */)
{
  closed_ = false;
  return 0;
}

int
PassSubscriberTask::close (unsigned long /* flags */)
{
  LOG_DEBUG ("Stopping subscription server...");
  closed_ = true;
  return 0;
}

int
PassSubscriberTask::svc (void)
{
  LOG_DEBUG ("PassSubscriberTask::svc() started");
  cfg_mgr_ = PassConfigurationManager::getInstance ();
  PASSSubscriberPortBindingService service (SOAP_C_UTFSTRING);
  
  LOG_DEBUG ("Binding SOAP server to port " << cfg_mgr_->getPassSubscriberPort () );
  
  SOAP_SOCKET res =
    service.bind (cfg_mgr_->getPassSubscriberInterface().c_str (), 
                  cfg_mgr_->getPassSubscriberPort (),
                  0);
                  
  if (res == SOAP_INVALID_SOCKET)
    {
      service.soap_print_fault (stderr);
      return 0;
    }
  
  int status = subscribe ();
  
  if (status != 0)
    {
      LOG_ERROR ("Error while subscribing...");
    }
  
  while (!closed_)
    {
      LOG_DEBUG ("Accepting connections...");
      
      // Server times out after 5 seconds (we ignore this error; we need
      // to time out so this thread can be cancelled).
      service.accept_timeout = 5;
      res = service.accept ();
      
      if (res == SOAP_INVALID_SOCKET && service.errnum != 0)
        {
          service.soap_print_fault (stderr);
          break;
        }
      else if (res == SOAP_INVALID_SOCKET && !service.errnum)
        {
          // TODO?
        }
      else
        {
          LOG_DEBUG ("Got a connection; serving it");
          status = service.serve ();
          
          if (status != SOAP_OK)
            {
              service.soap_print_fault (stderr);
            }
        }

#if 0 // TEST     
      sleep(2);
      testPublish();
#endif
    }
  
  LOG_DEBUG ("Unsubscribing...");
  status = unsubscribe ();
  
  if (status != 0)
    {
      LOG_ERROR ("Error while subscribing...");
    }
    
  service.soap_close_socket ();
  LOG_DEBUG ("Subscription thread terminating.");
  
{'fraction_non_alphanumeric': 0.12100980596703526, 'fraction_numerical': 0.005007302315877321, 'mean_word_length': 2.350104821802935, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, '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': '5550337', 'n_tokens_mistral': 1515, 'n_tokens_neox': 1487, 'n_words': 332}