//
// Client.cpp
//
// Library: Redis
// Package: Redis
// Module:  Client
//
// Implementation of the Client class.
//
// Copyright (c) 2015, Applied Informatics Software Engineering GmbH.
// and Contributors.
//
// SPDX-License-Identifier:	BSL-1.0
//


#include "Poco/Redis/Client.h"
#include "Poco/Redis/Exception.h"
#include "Poco/Redis/RedisNotifications.h"


namespace Poco {
namespace Redis {


Client::Client()
{
}


Client::Client(const std::string& hostAndPort):
	_address(hostAndPort)
{
	connect();
}


Client::Client(const std::string& host, int port):
	_address(host, port)
{
	connect();
}


Client::Client(const Net::SocketAddress& addrs):
	_address(addrs)
{
	connect();
}


Client::Client(const Net::StreamSocket& socket)
{
	connect(socket);
}


Client::~Client() = default;


void Client::connect()
{
	poco_assert(!_pInput);
	poco_assert(!_pOutput);

	try
	{
		_socket = Net::StreamSocket(_address);
		_pInput = std::make_unique<RedisInputStream>(_socket);
		_pOutput = std::make_unique<RedisOutputStream>(_socket);

		auto pNC = loadNC();
		if (pNC) pNC->postNotification(new RedisConnectNotification);
	}
	catch (const Exception& ex)
	{
		_pInput.reset();
		_pOutput.reset();

		auto pNC = loadNC();
		if (pNC) pNC->postNotification(new RedisErrorNotification(ex));
		throw;
	}
}


void Client::connect(const std::string& hostAndPort)
{
	_address = Net::SocketAddress(hostAndPort);
	connect();
}


void Client::connect(const std::string& host, int port)
{
	_address = Net::SocketAddress(host, port);
	connect();
}


void Client::connect(const Net::SocketAddress& addrs)
{
	_address = addrs;
	connect();
}


void Client::connect(const Timespan& timeout)
{
	poco_assert(!_pInput);
	poco_assert(!_pOutput);

	try
	{
		_socket = Net::StreamSocket();
		_socket.connect(_address, timeout);
		_pInput = std::make_unique<RedisInputStream>(_socket);
		_pOutput = std::make_unique<RedisOutputStream>(_socket);

		auto pNC = loadNC();
		if (pNC) pNC->postNotification(new RedisConnectNotification);
	}
	catch (const Exception& ex)
	{
		_pInput.reset();
		_pOutput.reset();

		auto pNC = loadNC();
		if (pNC) pNC->postNotification(new RedisErrorNotification(ex));
		throw;
	}
}


void Client::connect(const std::string& hostAndPort, const Timespan& timeout)
{
	_address = Net::SocketAddress(hostAndPort);
	connect(timeout);
}


void Client::connect(const std::string& host, int port, const Timespan& timeout)
{
	_address = Net::SocketAddress(host, port);
	connect(timeout);
}


void Client::connect(const Net::SocketAddress& addrs, const Timespan& timeout)
{
	_address = addrs;
	connect(timeout);
}


void Client::connect(const Poco::Net::StreamSocket& socket)
{
	poco_assert(!_pInput);
	poco_assert(!_pOutput);

	try
	{
		_address = socket.peerAddress();
		_socket = socket;
		_pInput = std::make_unique<RedisInputStream>(_socket);
		_pOutput = std::make_unique<RedisOutputStream>(_socket);

		auto pNC = loadNC();
		if (pNC) pNC->postNotification(new RedisConnectNotification);
	}
	catch (const Exception& ex)
	{
		_pInput.reset();
		_pOutput.reset();

		auto pNC = loadNC();
		if (pNC) pNC->postNotification(new RedisErrorNotification(ex));
		throw;
	}
}


void Client::disconnect()
{
	if (_pInput || _pOutput)
	{
		_pInput.reset();
		_pOutput.reset();

		_socket.close();

		auto pNC = loadNC();
		if (pNC) pNC->postNotification(new RedisDisconnectNotification);
	}
}


bool Client::isConnected() const
{
	return _pInput != nullptr;
}


void Client::writeCommand(const Array& command, bool doFlush)
{
	poco_assert(_pOutput);

	const std::string commandStr = command.toString();

	_pOutput->write(commandStr.c_str(), commandStr.length());
	if (doFlush) _pOutput->flush();
}


RedisType::Ptr Client::readReply()
{
	poco_assert(_pInput);

	const int c = _pInput->get();
	if (c == -1)
	{
		disconnect();
		throw RedisException("Lost connection to Redis server");
	}
	RedisType::Ptr result = RedisType::createRedisType(c);
	if (result.isNull())
	{
		throw RedisException("Invalid Redis type returned");
	}

	result->read(*_pInput);

	return result;
}


RedisType::Ptr Client::sendCommand(const Array& command)
{
	writeCommand(command, true);
	return readReply();
}


Array Client::sendCommands(const std::vector<Array>& commands)
{
	Array results;

	for (const auto& cmd : commands)
	{
		writeCommand(cmd, false);
	}
	_pOutput->flush();

	for (std::size_t i = 0; i < commands.size(); ++i)
	{
		results.addRedisType(readReply());
	}

	return results;
}


Client::NotificationCenterPtr Client::notificationCenter()
{
	std::call_once(_ncInitFlag, [this]()
	{
		storeNC(std::make_shared<AsyncNotificationCenter>());
	});
	return loadNC();
}


Client::NotificationCenterPtr Client::loadNC() const
{
#if POCO_HAVE_ATOMIC_SHARED_PTR
	return _pNC.load();
#else
	std::lock_guard<std::mutex> lock(_ncMutex);
	return _pNC;
#endif
}


void Client::storeNC(NotificationCenterPtr pNC)
{
#if POCO_HAVE_ATOMIC_SHARED_PTR
	_pNC.store(std::move(pNC));
#else
	std::lock_guard<std::mutex> lock(_ncMutex);
	_pNC = std::move(pNC);
#endif
}


} } // namespace Poco::Redis
