#include "Ipc.h"

#include <functional>

#include "ReadCallback.h"

std::shared_ptr<Ipc> Ipc::create(const char* reader_name, const char* writer_name) {
	struct Creator : public Ipc {
		Creator(const char* reader_name, const char* writer_name) : Ipc(reader_name, writer_name) {}
		~Creator() override = default;
	};
	return std::make_shared<Creator>(reader_name, writer_name);
}

Ipc::Ipc(const char* reader_name, const char* writer_name) noexcept
	: reader_name_(reader_name)
	, writer_name_(writer_name) {
	sender_ = std::make_unique<ipc::channel>(reader_name_.c_str(), ipc::sender);
	fprintf(stderr, "%s, %s\n", reader_name, writer_name);
}

Ipc::~Ipc() {
	stop();
}

bool Ipc::listen() {
	if (reciver_stop_.load(std::memory_order_acquire)) {
		return false;
	}

	reciver_stop_.store(false);

	std::weak_ptr<Ipc> wThis = shared_from_this();
	reciver_thread_ = std::move(std::thread(std::bind(&Ipc::doReciver, this, wThis)));
	return true;
}

void Ipc::stop() {
	if (!reciver_stop_.load(std::memory_order_acquire)) {
		return;
	}

	reciver_stop_.store(true, std::memory_order_release);

	sender_->disconnect();
	receiver_->disconnect();

	if (reciver_thread_.joinable()) {
		reciver_thread_.join();
	}
}

bool Ipc::send(const char* data, unsigned int size) {
	if (!sender_) {
		return false;
	}

	return sender_->send(data, size);
}

void Ipc::reConnect() {
    sender_->reconnect(ipc::sender);
    if (receiver_) {
        receiver_->disconnect();
        receiver_->reconnect(ipc::receiver);
    }
}

void Ipc::registReadCallback(const std::shared_ptr<IReaderCallback>& reader) {
	std::lock_guard<std::mutex> lock(mutex_);

	auto itor = std::find(reader_callbacks_.begin(), reader_callbacks_.end(), reader);
	if (reader_callbacks_.end() != itor) {
		return;
	}
	reader_callbacks_.emplace_back(reader);
}

void Ipc::unregistReadCallback(const std::shared_ptr<IReaderCallback>& reader) {
	std::lock_guard<std::mutex> lock(mutex_);

	auto itor = std::find(reader_callbacks_.begin(), reader_callbacks_.end(), reader);
	if (reader_callbacks_.end() != itor) {
		return;
	}
	reader_callbacks_.erase(itor);
}

void Ipc::doReciver(std::weak_ptr<Ipc> wThis) {
	std::shared_ptr<Ipc> self(wThis.lock());
	if (!self) {
		return;
	}

	receiver_ = std::make_unique<ipc::channel>(writer_name_.c_str(), ipc::receiver);
	while (!reciver_stop_.load(std::memory_order_acquire)) {
		ipc::buff_t buffer = receiver_->recv();
		if (buffer.empty()) {
			break;
		}
		const char* data = buffer.get<const char*>();
		onReciveer(data, static_cast<unsigned int>(buffer.size()));
	}
}

void Ipc::onReciveer(const char* buffer, unsigned int size) {
	std::lock_guard<std::mutex> lock(mutex_);
	for (auto& read : reader_callbacks_) {
		read->onRead(buffer, size);
	}
}
