Bi-directional pipe as thread queue

BidiPipe.h

#ifndef __SRV_BIDIPIPE_H__
#define __SRV_BIDIPIPE_H__

#include "Util.h"

#ifndef ARRAYLEN
#define ARRAYLEN(array)     (sizeof(array)/sizeof((array)[0]))
#endif // ARRAYLEN

class BidiPipe {
public:
	enum Direction {
		Undefined,
		Upstream,
		Downstream,
	};

protected:
	int pipe_upstream[2];		// downstream -> upstream
	int pipe_downstream[2];		// upstream -> downstream
	bool isOpen;
	enum Direction direction;

public:
	BidiPipe()
	{
		isOpen = false;
		direction = Undefined;
		for (unsigned int i = 0; i < ARRAYLEN(pipe_upstream); i++) {
			pipe_upstream[i] = -1;
			pipe_downstream[i] = -1;
		}
	}
	BidiPipe(const BidiPipe& old) {
		for (unsigned int i = 0; i < ARRAYLEN(pipe_upstream); i++) {
			pipe_upstream[i] = old.pipe_upstream[i];
			pipe_downstream[i] = old.pipe_downstream[i];
		}
		isOpen = old.isOpen;
		direction = old.direction;
	}
	~BidiPipe()
	{
		close();
	}

	void close()
	{
		for (unsigned int i = 0; i < ARRAYLEN(pipe_upstream); i++) {
			if (pipe_upstream[i] >= 0) {
				::close(pipe_upstream[i]);
				pipe_upstream[i] = -1;
			}
			if (pipe_downstream[i] >= 0) {
				::close(pipe_downstream[i]);
				pipe_downstream[i] = -1;
			}
		}

		isOpen = false;
	}

	bool open()
	{
		if (pipe(pipe_upstream))
			return false;
		if (pipe(pipe_downstream))
			return false;

		for (unsigned int i = 0; i < ARRAYLEN(pipe_upstream); i++) {
			if (!set_nonblock(pipe_upstream[i], nullptr))
				return false;
			if (!set_nonblock(pipe_downstream[i], nullptr))
				return false;
		}

		isOpen = true;
		return true;
	}

	bool setDirection(enum Direction directionIn)
	{
		if (!isOpen || direction != Undefined)
			return false;

		if (directionIn == Upstream) {
			::close(pipe_upstream[1]);
			pipe_upstream[1] = -1;

			::close(pipe_downstream[0]);
			pipe_downstream[0] = -1;
		} else {
			::close(pipe_upstream[0]);
			pipe_upstream[0] = -1;

			::close(pipe_downstream[1]);
			pipe_downstream[1] = -1;
		}

		direction = directionIn;
		return true;
	}

	int readFd()
	{
		if (direction == Upstream)
			return pipe_upstream[0];
		else if (direction == Downstream)
			return pipe_downstream[0];
		else
			return -1;
	}

	int writeFd()
	{
		if (direction == Upstream)
			return pipe_downstream[1];
		else if (direction == Downstream)
			return pipe_upstream[1];
		else
			return -1;
	}

	ssize_t read(void *buf, size_t count)
	{
		int fd = readFd();
		if (fd < 0)
			return -1;

		return ::read(fd, buf, count);
	}

	ssize_t write(const void *buf, size_t count)
	{
		int fd = writeFd();
		if (fd < 0)
			return -1;

		return ::write(fd, buf, count);
	}

	bool readCmd(uint32_t& res)
	{
		ssize_t bread = read(&res, sizeof(res));
		return (bread == sizeof(res));
	}

	bool writeCmd(uint32_t v)
	{
		ssize_t bwritten = write(&v, sizeof(v));
		return (bwritten == sizeof(v));
	}
};

#endif // __SRV_BIDIPIPE_H__