text: // See the file "COPYING" in the main distribution directory for copyright.
  
  #include <broker/data.hh>
  
  #include "util.h"
  #include "threading/SerialTypes.h"
  
  #include "Manager.h"
  #include "WriterBackend.h"
  #include "WriterFrontend.h"
  
  // Messages sent from backend to frontend (i.e., "OutputMessages").
  
  using zeek::threading::Value;
  using zeek::threading::Field;
  
  namespace zeek::logging {
  
  class RotationFinishedMessage final : public threading::OutputMessage<WriterFrontend>
  {
  public:
  	RotationFinishedMessage(WriterFrontend* writer, const char* new_name, const char* old_name,
  				double open, double close, bool success, bool terminating)
  		: threading::OutputMessage<WriterFrontend>("RotationFinished", writer),
  		new_name(util::copy_string(new_name)), old_name(util::copy_string(old_name)), open(open),
  		close(close), success(success), terminating(terminating)	{ }
  
  	~RotationFinishedMessage() override
  		{
  		delete [] new_name;
  		delete [] old_name;
  		}
  
  	bool Process() override
  		{
  		return log_mgr->FinishedRotation(Object(), new_name, old_name, open, close, success, terminating);
  		}
  
  private:
          const char* new_name;
          const char* old_name;
          double open;
          double close;
  	bool success;
          bool terminating;
  };
  
  class FlushWriteBufferMessage final : public threading::OutputMessage<WriterFrontend>
  {
  public:
  	FlushWriteBufferMessage(WriterFrontend* writer)
  		: threading::OutputMessage<WriterFrontend>("FlushWriteBuffer", writer)	{}
  
  	bool Process() override	{ Object()->FlushWriteBuffer(); return true; }
  };
  
  class DisableMessage final : public threading::OutputMessage<WriterFrontend>
  {
  public:
  	DisableMessage(WriterFrontend* writer)
  		: threading::OutputMessage<WriterFrontend>("Disable", writer)	{}
  
  	bool Process() override	{ Object()->SetDisable(); return true; }
  };
  
  // Backend methods.
  
  broker::data WriterBackend::WriterInfo::ToBroker() const
  	{
  	auto t = broker::table();
  
  	for ( config_map::const_iterator i = config.begin(); i != config.end(); ++i )
  		{
  		auto key = std::string(i->first);
  		auto value = std::string(i->second);
  		t.insert(std::make_pair(key, value));
  		}
  
  	auto bppf = post_proc_func ? post_proc_func : "";
  
  	return broker::vector({path, rotation_base, rotation_interval, network_time, std::move(t), bppf});
  	}
  
  bool WriterBackend::WriterInfo::FromBroker(broker::data d)
  	{
  	if ( ! caf::holds_alternative<broker::vector>(d) )
  		return false;
  
  	auto v = caf::get<broker::vector>(d);
  	auto bpath = caf::get_if<std::string>(&v[0]);
  	auto brotation_base = caf::get_if<double>(&v[1]);
  	auto brotation_interval = caf::get_if<double>(&v[2]);
  	auto bnetwork_time = caf::get_if<double>(&v[3]);
  	auto bconfig = caf::get_if<broker::table>(&v[4]);
  	auto bppf = caf::get_if<std::string>(&v[5]);
  
  	if ( ! (bpath && brotation_base && brotation_interval && bnetwork_time && bconfig && bppf) )
  		return false;
  
  	path = util::copy_string(bpath->c_str());
  	post_proc_func = util::copy_string(bppf->c_str());
  	rotation_base = *brotation_base;
  	rotation_interval = *brotation_interval;
  	network_time = *bnetwork_time;
  
  	for ( auto i : *bconfig )
  		{
  		auto k = caf::get_if<std::string>(&i.first);
  		auto v = caf::get_if<std::string>(&i.second);
  
  		if ( ! (k && v) )
  			return false;
  
  		auto p = std::make_pair(util::copy_string(k->c_str()), util::copy_string(v->c_str()));
  		config.insert(p);
  		}
  
  	return true;
  	}
  
  WriterBackend::WriterBackend(WriterFrontend* arg_frontend) : MsgThread()
  	{
  	num_fields = 0;
  	fields = nullptr;
  	buffering = true;
  	frontend = arg_frontend;
  	info = new WriterInfo(frontend->Info());
  	rotation_counter = 0;
  
  	SetName(frontend->Name());
  	}
  
  WriterBackend::~WriterBackend()
  	{
  	if ( fields )
  		{
  		for(int i = 0; i < num_fields; ++i)
  			delete fields[i];
  
  		delete [] fields;
  		}
  
  	delete info;
  	}
  
  void WriterBackend::DeleteVals(int num_writes, Value*** vals)
  	{
  	for ( int j = 0; j < num_writes; ++j )
  		{
  		// Note this code is duplicated in Manager::DeleteVals().
  		for ( int i = 0; i < num_fields; i++ )
  			delete vals[j][i];
  
  		delete [] vals[j];
  		}
  
  	delete [] vals;
  	}
  
  bool WriterBackend::FinishedRotation(const char* new_name, const char* old_name,
  				     double open, double close, bool terminating)
  	{
  	--rotation_counter;
  	SendOut(new RotationFinishedMessage(frontend, new_name, old_name, open, close, true, terminating));
  	return true;
  	}
  
  bool WriterBackend::FinishedRotation()
  	{
  	--rotation_counter;
  	SendOut(new RotationFinishedMessage(frontend, nullptr, nullptr, 0, 0, false, false));
  	return true;
  	}
  
  void WriterBackend::DisableFrontend()
  	{
  	SendOut(new DisableMessage(frontend));
  	}
  
  bool WriterBackend::Init(int arg_num_fields, const Field* const* arg_fields)
  	{
  	SetOSName(Fmt("zk.%s", Name()));
  	num_fields = arg_num_fields;
  	fields = arg_fields;
  
  	if ( Failed() )
  		return true;
  
  	if ( ! DoInit(*info, arg_num_fields, arg_fields) )
  		{
  		DisableFrontend();
  		return false;
  		}
  
  	return true;
  	}
  
  bool WriterBackend::Write(int arg_num_fields, int num_writes, Value*** vals)
  	{
  	// Double-check that the arguments match. If we get this from remote,
  	// something might be mixed up.
  	if ( num_fields != arg_num_fields )
  		{
  
  #ifdef DEBUG
  		const char* msg = Fmt("Number of fields don't match in WriterBackend::Write() (%d vs. %d)",
  				      arg_num_fields, num_fields);
  		Debug(DBG_LOGGING, msg);
  #endif
  
  		DeleteVals(num_writes, vals);
  		DisableFrontend();
  		return false;
  		}
  
  	// Double-check all the types match.
  	for ( int j = 0; j < num_writes; j++ )
  		{
  		for ( int i = 0; i < num_fields; ++i )
  			{
  			if ( vals[j][i]->type != fields[i]->type )
  				{
  #ifdef DEBUG
  				const char* msg = Fmt("Field #%d type doesn't match in WriterBackend::Write() (%d vs. %d)",
  						      i, vals[j][i]->type, fields[i]->type);
  				Debug(DBG_LOGGING, msg);
  #endif
  				DisableFrontend();
  				DeleteVals(num_writes, vals);
  				return false;
  				}
  			}
  		}
  
  	bool success = true;
  
  	if ( ! Failed() )
  		{
  		for ( int j = 0; j < num_writes; j++ )
  			{
  			success = DoWrite(num_fields, fields, vals[j]);
  
  			if ( ! success )
  				break;
  			}
  		}
  
  	DeleteVals(num_writes, vals);
  
  	if ( ! success )
  		DisableFrontend();
  
  	return success;
  	}
  
  bool WriterBackend::SetBuf(bool enabled)
  	{
  	if ( enabled == buffering )
  		// No change.
  		return true;
  
  	if ( Failed() )
  		return true;
  
  	buffering = enabled;
  
  	if ( ! DoSetBuf(enabled) )
  		{
  		DisableFrontend();
  		return false;
  		}
  
  	return true;
  	}
  
  bool WriterBackend::Rotate(const char* rotated_path, double open,
  			   double close, bool terminating)
  	{
  	if ( Failed() )
  		return true;
  
  	rotation_counter = 1;
  
  	if ( ! DoRotate(rotated_path, open, close, terminating) )
  		{
  		DisableFrontend();
  		return false;
  		}
  
  	// Insurance against broken writers.
  	if ( rotation_counter > 0 )
  		InternalError(Fmt("writer %s did not call FinishedRotation() in DoRotation()", Name()));
  
  	if ( rotation_counter < 0 )
  		InternalError(Fmt("writer %s called FinishedRotation() more than once in DoRotation()", Name()));
  
  	return true;
  	}
  
  bool WriterBackend::Flush(double network_time)
  	{
  	if ( Failed() )
  		return true;
  
  	if ( ! DoFlush(network_time) )
  		{
  		DisableFrontend();
  		return false;
  		}
  
  	return true;
  	}
  
  bool WriterBackend::OnFinish(double network_time)
  	{
  	if ( Failed() )
  		return true;
  
  	return DoFinish(network_time);
  	}
  
  bool WriterBackend::OnHeartbeat(double network_time, double current_time)
  	{
  	if ( Failed() )
  		return true;
  
  	SendOut(new FlushWriteBufferMessage(frontend));
  	return DoHeartbeat(network_time, current_time);
  	}
  
  } // namespace zeek::logging
  