librsync job reader & writer

rsynclib.h

/*
 * Copyright 2015 Rackspace, Inc.
 *
 * Licensed 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.
 */

#pragma once

#include <librsync.h>

#include <array>
#include <cstring>

#if defined(_WIN32)
#include <ciso646>
#include <io.h>
#else
#include <unistd.h>
#define _read read
#define _write write
#endif

namespace deuterium
{

const size_t iobufsize = 16 * 1024;

struct rsync_base
{
	rsync_base(int fd, rs_job_t* job) :
		fd_(fd),
		job_(job),
		rbuf_()
	{}

	char const* message() const
	{
		return rs_strerror(r_);
	}

	~rsync_base()
	{
		rs_job_free(job_);
	}

protected:
	int fd_;
	rs_job_t* job_;
	rs_buffers_t rbuf_;
	rs_result r_;
};

struct rsync_writer : rsync_base
{
	rsync_writer(int fd, rs_job_t* job) :
		rsync_base(fd, job)
	{
		reset_buffer();
	}

	bool update(char* p, size_t sz)
	{
		rbuf_.next_in = p;
		rbuf_.avail_in = sz;
		continue_job();

		return r_ == RS_BLOCKED or r_ == RS_DONE;
	}

	bool final()
	{
		rbuf_.eof_in = true;
		continue_job();

		while (r_ == RS_BLOCKED and rbuf_.avail_out != buf_.size() and
		    do_flush())
			r_ = rs_job_iter(job_, &rbuf_);

		return r_ == RS_DONE and do_flush();
	}

private:
	void continue_job()
	{
		for (;;)
		{
			r_ = rs_job_iter(job_, &rbuf_);

			// output buffer out-of-space
			if (r_ == RS_BLOCKED and rbuf_.avail_in != 0 and
			    do_flush())
				continue;

			break;
		}
	}

	void reset_buffer()
	{
		rbuf_.next_out = buf_.data();
		rbuf_.avail_out = buf_.size();
	}

	bool do_flush()
	{
		auto blen = int(buf_.size() - rbuf_.avail_out);
		bool ok = _write(fd_, buf_.data(), blen) == blen;

		if (ok)
			reset_buffer();
		else
			r_ = RS_IO_ERROR;

		return ok;
	}

	std::array<char, iobufsize> buf_;
};

struct rsync_reader : rsync_base
{
	rsync_reader(int fd, rs_job_t* job) :
		rsync_base(fd, job)
	{
		reset_buffer();
	}

	bool apply(char* p, size_t sz)
	{
		rbuf_.next_out = p;
		rbuf_.avail_out = sz;
		continue_job();

		return r_ == RS_BLOCKED or r_ == RS_DONE;
	}

	size_t pending_bytes() const
	{
		return  rbuf_.avail_out;
	}

	bool load_all()
	{
		do
			continue_job();
		while (r_ == RS_BLOCKED);

		return r_ == RS_DONE;
	}

private:
	void continue_job()
	{
		// enough inactive space to load data
		if ((buf_.size() - rbuf_.avail_in) >= iobufsize)
		{
			// not enough unused space
			if ((buf_.size() - (rbuf_.next_in - buf_.data() +
			    rbuf_.avail_in)) < iobufsize)
			{
				memmove(buf_.data(), rbuf_.next_in,
				    rbuf_.avail_in);
				reset_buffer();
			}

			auto n = _read(fd_, rbuf_.next_in +
			    rbuf_.avail_in, iobufsize);

			if (n == -1)
			{
				r_ = RS_IO_ERROR;
				return;
			}
			else if (n == 0)
				rbuf_.eof_in = true;
			else
				rbuf_.avail_in += n;
		}

		r_ = rs_job_iter(job_, &rbuf_);
	}

	void reset_buffer()
	{
		rbuf_.next_in = buf_.data();
	}

	// make enough space to load pending data + iobufsize
	// hope byte-oriented read(2) still likes unaligned address
	std::array<char, iobufsize * 2> buf_;
};

}

#undef _write
#undef _read