/*

 * Licensed to the Apache Software Foundation (ASF) under one or more

 * contributor license agreements.  See the NOTICE file distributed with

 * this work for additional information regarding copyright ownership.

 * The ASF licenses this file to You under the Apache License, Version 2.0

 * (the "License"); you may not use this file except in compliance with

 * the License.  You may obtain a copy of the License at

 *

 *      http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

#define __STDC_CONSTANT_MACROS

#include <log4cxx/net/socketappenderskeleton.h>

#include <log4cxx/helpers/loglog.h>

#include <log4cxx/helpers/charsetencoder.h>

#include <log4cxx/helpers/optionconverter.h>

#include <log4cxx/helpers/stringhelper.h>

#include <log4cxx/spi/loggingevent.h>

#include <log4cxx/helpers/threadutility.h>

#include <log4cxx/helpers/transcoder.h>

#include <log4cxx/helpers/bytearrayoutputstream.h>

#include <log4cxx/helpers/outputstreamwriter.h>

#include <log4cxx/helpers/socketoutputstream.h>

#include <log4cxx/helpers/threadutility.h>

#include <log4cxx/private/socketappenderskeleton_priv.h>

#include <functional>

#include <chrono>

using namespace LOG4CXX_NS;

using namespace LOG4CXX_NS::helpers;

using namespace LOG4CXX_NS::net;

#define _priv static_cast<SocketAppenderSkeletonPriv*>(m_priv.get())

SocketAppenderSkeleton::SocketAppenderSkeleton(int defaultPort, int reconnectionDelay)

  : AppenderSkeleton(std::make_unique<SocketAppenderSkeletonPriv>(defaultPort, reconnectionDelay))

{

}

#if LOG4CXX_ABI_VERSION <= 15

SocketAppenderSkeleton::SocketAppenderSkeleton(helpers::InetAddressPtr address, int port, int reconnectionDelay)

#else

SocketAppenderSkeleton::SocketAppenderSkeleton(const helpers::InetAddressPtr& address, int port, int reconnectionDelay)

#endif

  : AppenderSkeleton(std::make_unique<SocketAppenderSkeletonPriv>(address, port, reconnectionDelay))

{

}

SocketAppenderSkeleton::SocketAppenderSkeleton(const LogString& host, int port, int reconnectionDelay)

  : AppenderSkeleton(std::make_unique<SocketAppenderSkeletonPriv>(host, port, reconnectionDelay))

{

}

SocketAppenderSkeleton::SocketAppenderSkeleton(std::unique_ptr<SocketAppenderSkeletonPriv> priv)

  :  AppenderSkeleton (std::move(priv))

{

}

SocketAppenderSkeleton::~SocketAppenderSkeleton()

{

}

void SocketAppenderSkeleton::activateOptions( LOG4CXX_ACTIVATE_OPTIONS_FORMAL_PARAMETERS )

{

  _priv->connect();

}

void SocketAppenderSkeleton::close()

{

  if (_priv->setClosed())

    _priv->close();

}

void SocketAppenderSkeleton::SocketAppenderSkeletonPriv::connect()

{

  if (this->address == 0)

  {

    LogLog::error(LogString(LOG4CXX_STR("No remote host is set for Appender named \"")) +

      this->name + LOG4CXX_STR("\"."));

  }

  else

  {

    this->close();

    try

    {

      if (LogLog::isDebugEnabled())

      {

        LogString msg(LOG4CXX_STR("Connecting to [")

          + this->address->toString() + LOG4CXX_STR(":"));

        StringHelper::toString(this->port, msg);

        msg += LOG4CXX_STR("].");

        LogLog::debug(msg);

      }

      this->setOutputSink(Socket::create(this->address, this->port, this->socketSubclass));

    }

    catch (SocketException& e)

    {

      LogString msg = LOG4CXX_STR("Could not connect to [")

        + this->address->toString() + LOG4CXX_STR(":");

      StringHelper::toString(this->port, msg);

      msg += LOG4CXX_STR("].");

      this->fireConnector(); // fire the connector thread

      LogLog::warn(msg, e);

    }

  }

}

void SocketAppenderSkeleton::setOption(const LogString& option, const LogString& value)

{

  if (StringHelper::equalsIgnoreCase(option, LOG4CXX_STR("REMOTEHOST"), LOG4CXX_STR("remotehost")))

  {

    setRemoteHost(value);

  }

  else if (StringHelper::equalsIgnoreCase(option, LOG4CXX_STR("PORT"), LOG4CXX_STR("port")))

  {

    setPort(OptionConverter::toInt(value, getDefaultPort()));

  }

  else if (StringHelper::equalsIgnoreCase(option, LOG4CXX_STR("LOCATIONINFO"), LOG4CXX_STR("locationinfo")))

  {

    setLocationInfo(OptionConverter::toBoolean(value, false));

  }

  else if (StringHelper::equalsIgnoreCase(option, LOG4CXX_STR("RECONNECTIONDELAY"), LOG4CXX_STR("reconnectiondelay")))

  {

    setReconnectionDelay(OptionConverter::toInt(value, getDefaultDelay()));

  }

#if 15 < LOG4CXX_ABI_VERSION

  else if (StringHelper::equalsIgnoreCase(option, LOG4CXX_STR("SOCKETSUBCLASS"), LOG4CXX_STR("socketsubclass")))

  {

    setSocketSubclass(value);

  }

#endif

  else

  {

    AppenderSkeleton::setOption(option, value);

  }

}

void SocketAppenderSkeleton::SocketAppenderSkeletonPriv::fireConnector()

{

  std::lock_guard<std::recursive_mutex> lock(this->mutex);

  if (this->taskName.empty())

  {

    this->taskName = this->name + LOG4CXX_STR(":")

      + this->address->toString() + LOG4CXX_STR(":");

    StringHelper::toString(this->port, this->taskName);

  }

  auto taskManager = ThreadUtility::instancePtr();

  if (!taskManager->value().hasPeriodicTask(this->taskName))

  {

    if (LogLog::isDebugEnabled())

    {

      LogString msg(LOG4CXX_STR("Waiting "));

      StringHelper::toString(this->reconnectionDelay, msg);

      msg += LOG4CXX_STR(" ms before retrying [")

        + this->address->toString() + LOG4CXX_STR(":");

      StringHelper::toString(this->port, msg);

      msg += LOG4CXX_STR("].");

      LogLog::debug(msg);

    }

    taskManager->value().addPeriodicTask(this->taskName

      , std::bind(&SocketAppenderSkeleton::SocketAppenderSkeletonPriv::retryConnect, this)

      , std::chrono::milliseconds(this->reconnectionDelay)

      );

  }

  this->taskManager = taskManager;

}

#if LOG4CXX_ABI_VERSION <= 15

void SocketAppenderSkeleton::fireConnector()

{

  _priv->fireConnector();

}

#endif

void SocketAppenderSkeleton::SocketAppenderSkeletonPriv::retryConnect()

{

  if (this->closed)

  {

    if (auto pManager = this->taskManager.lock())

      pManager->value().removePeriodicTask(this->taskName);

  }

  else

  {

    try

    {

      if (LogLog::isDebugEnabled())

      {

        LogString msg(LOG4CXX_STR("Attempting connection to [")

          + this->address->toString() + LOG4CXX_STR(":"));

        StringHelper::toString(this->port, msg);

        msg += LOG4CXX_STR("].");

        LogLog::debug(msg);

      }

      this->setOutputSink(Socket::create(this->address, this->port, this->socketSubclass));

      if (LogLog::isDebugEnabled())

      {

        LogString msg(LOG4CXX_STR("Connection established to [")

          + this->address->toString() + LOG4CXX_STR(":"));

        StringHelper::toString(this->port, msg);

        msg += LOG4CXX_STR("].");

        LogLog::debug(msg);

      }

      if (auto pManager = this->taskManager.lock())

        pManager->value().removePeriodicTask(this->taskName);

      return;

    }

    catch (ConnectException& e)

    {

      LogLog::warn(LOG4CXX_STR("Remote host ")

        + this->address->toString()

        + LOG4CXX_STR(" refused connection."), e);

    }

    catch (IOException& e)

    {

      LogString msg(LOG4CXX_STR("Could not connect to [")

        + this->address->toString() + LOG4CXX_STR(":"));

      StringHelper::toString(this->port, msg);

      msg += LOG4CXX_STR("].");

      LogLog::warn(msg, e);

    }

    if (this->reconnectionDelay > 0)

    {

      if (LogLog::isDebugEnabled())

      {

        LogString msg(LOG4CXX_STR("Waiting "));

        StringHelper::toString(this->reconnectionDelay, msg);

        msg += LOG4CXX_STR(" ms before retrying [")

          + this->address->toString() + LOG4CXX_STR(":");

        StringHelper::toString(this->port, msg);

        msg += LOG4CXX_STR("].");

        LogLog::debug(msg);

      }

    }

  }

}

void SocketAppenderSkeleton::SocketAppenderSkeletonPriv::setOutputSink(const SocketPtr& socket)

{

  OutputStreamPtr os = std::make_shared<SocketOutputStream>(socket);

  auto charset = CharsetEncoder::getUTF8Encoder();

  this->outputSink = std::make_shared<OutputStreamWriter>(os, charset);

}

void SocketAppenderSkeleton::SocketAppenderSkeletonPriv::close()

{

  if (this->taskName.empty())

    ;

  else if (auto pManager = this->taskManager.lock())

    pManager->value().removePeriodicTask(this->taskName);

  if (this->outputSink)

  {

    try

    {

      this->outputSink->close();

      this->outputSink.reset();

    }

    catch (std::exception&)

    {

    }

  }

}

void SocketAppenderSkeleton::setRemoteHost(const LogString& host)

{

  _priv->address = helpers::InetAddress::getByName(host);

  _priv->remoteHost.assign(host);

}

const LogString& SocketAppenderSkeleton::getRemoteHost() const

{

  return _priv->remoteHost;

}

void SocketAppenderSkeleton::setPort(int port1)

{

  _priv->port = port1;

}

int SocketAppenderSkeleton::getPort() const

{

  return _priv->port;

}

void SocketAppenderSkeleton::setLocationInfo(bool locationInfo1)

{

  _priv->locationInfo = locationInfo1;

}

bool SocketAppenderSkeleton::getLocationInfo() const

{

  return _priv->locationInfo;

}

void SocketAppenderSkeleton::setReconnectionDelay(int reconnectionDelay1)

{

  _priv->reconnectionDelay = reconnectionDelay1;

  if (_priv->taskName.empty())

    return;

  auto pManager = _priv->taskManager.lock();

  if (pManager && pManager->value().hasPeriodicTask(_priv->taskName))

  {

    pManager->value().removePeriodicTask(_priv->taskName);

    pManager->value().addPeriodicTask(_priv->taskName

      , std::bind(&SocketAppenderSkeleton::SocketAppenderSkeletonPriv::retryConnect, _priv)

      , std::chrono::milliseconds(_priv->reconnectionDelay)

      );

  }

}

int SocketAppenderSkeleton::getReconnectionDelay() const

{

  return _priv->reconnectionDelay;

}

#if 15 < LOG4CXX_ABI_VERSION

void SocketAppenderSkeleton::setSocketSubclass(const LogString& newValue)

{

  _priv->socketSubclass = newValue;

}

const LogString& SocketAppenderSkeleton::getSocketSubclass() const

{

  return _priv->socketSubclass;

}

#endif