client-link.5c

/*
 * Copyright © 2012 Keith Packard <keithp@keithp.com>
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; version 2 of the License.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License along
 * with this program; if not, write to the Free Software Foundation, Inc.,
 * 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
 */

autoload RR;
autoload RR::Lex;
autoload RR::Send;
autoload Client;
autoload Client::Net;
autoload Mutex;
autoload List;

extend namespace Client {
	public namespace Link {

		public typedef List::list_t + struct {
			string[*]	reply;
		} message_struct;

		public typedef *message_struct message_t;

		public exception link_error(string);

		typedef struct {
			file		f;
			string		host;
			int		port;
			List::list_t	replies;
			semaphore	replies_sem;
			List::list_t	notices;
			semaphore	notices_sem;
			bool		closing;
			Mutex::mutex	command_lock;
			Mutex::mutex	notice_lock;
			Mutex::mutex	reply_lock;
			thread		reader;
		} link_struct;

		public typedef *link_struct link_t;

		public exception link_closing();

		void read_replies(link_t l) {
			try {
				for (;;) {
					message_t m = &(message_struct) { .reply = RR::Lex::recv(l->f) };
					if (dim(m->reply) == 0)
						continue;
					twixt (Mutex::acquire(l->reply_lock); Mutex::release(l->reply_lock)) {
						if (m->reply[0] == "NOTICE") {
							List::append(m, &l->notices);
							Semaphore::signal(l->notices_sem);
						} else {
							List::append(m, &l->replies);
							Semaphore::signal(l->replies_sem);
						}
					}
				}
			} catch Thread::signal(int sig) {
			} catch File::io_eof(file f) {
				l->closing = true;
			} catch File::io_error(string reason, File::error_type error, file f) {
				l->closing = true;
			}
			while (Semaphore::count(l->notices_sem) < 0)
				Semaphore::signal(l->notices_sem);
			while (Semaphore::count(l->replies_sem) < 0)
				Semaphore::signal(l->replies_sem);
		}

		public message_t command(link_t l, string format, poly args...) {
			message_t reply;
			twixt (Mutex::acquire(l->command_lock); Mutex::release(l->command_lock)) {
				try {
					RR::Send::send(l->f, format, args...);
					File::flush(l->f);
				} catch File::io_error(string reason, File::error_type error, file f) {
					raise link_error(sprintf("I/O error on link: %s", reason));
				}
				Semaphore::wait(l->replies_sem);
				twixt(Mutex::acquire(l->reply_lock); Mutex::release(l->reply_lock)) {
					if (l->closing)
						raise link_closing();
					reply = List::first(&l->replies);
					List::remove(reply);
				}
			}
			return reply;
		}

		public message_t notice(link_t l) {
			message_t notice;
			twixt (Mutex::acquire(l->notice_lock); Mutex::release(l->notice_lock)) {
				Semaphore::wait(l->notices_sem);
				twixt(Mutex::acquire(l->reply_lock); Mutex::release(l->reply_lock)) {
					if (l->closing)
						raise link_closing();
					notice = List::first(&l->notices);
					List::remove(notice);
				}
			}
			return notice;
		}

		public link_t new (string host, int port) {
			link_t	l = &(link_struct) {
				.host = host,
				.port = port,
				.closing = false,
				.command_lock = Mutex::new(),
				.notice_lock = Mutex::new(),
				.reply_lock = Mutex::new(),
				.replies_sem = Semaphore::new(),
				.notices_sem = Semaphore::new(),
			};

			List::init(&l->replies);
			List::init(&l->notices);

			l->f = Net::connect (l->host, l->port);
			l->reader = fork read_replies(l);
			return l;
		}

		public void close(link_t l) {
			l->closing = true;
			Thread::send_signal(l->reader, 1);
			Thread::join(l->reader);
			File::close(l->f);
		}
	}
}