//Prevayler(TM) - The Free-Software Prevalence Layer.
//Copyright (C) 2001-2003 Klaus Wuestefeld
//This library 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.

package org.prevayler.implementation.publishing;

import java.io.IOException;
import java.util.Date;

import org.prevayler.Clock;
import org.prevayler.Transaction;
import org.prevayler.foundation.Cool;
import org.prevayler.foundation.Turn;
import org.prevayler.implementation.clock.PausableClock;
import org.prevayler.implementation.journal.Journal;
import org.prevayler.implementation.publishing.censorship.*;

public class CentralPublisher extends AbstractPublisher {

	private final PausableClock _pausableClock;
	private final TransactionCensor _censor;
	private final Journal _journal;

	private final Object _pendingSubscriptionMonitor = new Object();
	private volatile int _pendingPublications = 0;
	private final Object _pendingPublicationsMonitor = new Object();

	private Turn _nextTurn = Turn.first();
	private final Object _nextTurnMonitor = new Object();

	private int _transactionsToJournal = 0;
	private final Object _transactionsToJournalMonitor = new Object();


	public CentralPublisher(Clock clock, TransactionCensor censor, Journal journal) {
		super(new PausableClock(clock));
		_pausableClock = (PausableClock)_clock; //This is just to avoid casting the inherited _clock every time.

		_censor = censor;
		_journal = journal;
	}


	public void publish(Transaction transaction) {
		synchronized (_pendingSubscriptionMonitor) {  //Blocks all new publications until the subscription is over.
			synchronized (_pendingPublicationsMonitor) {
				if (_pendingPublications == 0) _pausableClock.pause();
				_pendingPublications++;
			}
		}

		try {
			publishWithoutWorryingAboutNewSubscriptions(transaction);  // Suggestions for a better method name are welcome.  :)
		} finally {
			synchronized (_pendingPublicationsMonitor) {
				_pendingPublications--;
				if (_pendingPublications == 0) _pausableClock.resume();
			}
		}
	}


	private void publishWithoutWorryingAboutNewSubscriptions(Transaction transaction) {
		Turn myTurn = nextTurn();

		Date executionTime = realTime(myTurn);  //TODO realTime() and approve in the same turn.
		approve(transaction, executionTime, myTurn);
		journal(transaction, myTurn, executionTime);
		notifySubscribers(transaction, executionTime, myTurn);
	}


	private Turn nextTurn() {
		synchronized (_nextTurnMonitor) {
			Turn result = _nextTurn;
			_nextTurn = _nextTurn.next();
			return result;
		}
	}


	private Date realTime(Turn myTurn) {
		try {
			myTurn.start();
			return _pausableClock.realTime();
		} finally {	myTurn.end(); }
	}


	private void approve(Transaction transaction, Date executionTime, Turn myTurn) throws RuntimeException, Error {
		try {
			myTurn.start();
			_censor.approve(transaction, executionTime);
			synchronized (_transactionsToJournalMonitor) { _transactionsToJournal++; }
			myTurn.end();
		} catch (RuntimeException r) { dealWithError(myTurn); throw r;
		} catch (Error e) {	dealWithError(myTurn); throw e; }
	}

	
	private void dealWithError(Turn myTurn) {
		synchronized(_transactionsToJournalMonitor) {
			while (_transactionsToJournal != 0) Cool.wait(_transactionsToJournalMonitor);
		}
		myTurn.alwaysSkip();
	}

	
	private void journal(Transaction transaction, Turn myTurn, Date executionTime) {
		_journal.append(transaction, executionTime, myTurn);
		synchronized (_transactionsToJournalMonitor) {
			_transactionsToJournal--;
			if (_transactionsToJournal == 0) _transactionsToJournalMonitor.notify(); 
		}
	}

	
	private void notifySubscribers(Transaction transaction, Date executionTime, Turn myTurn) {
		try {
			myTurn.start();
			_pausableClock.advanceTo(executionTime);
			notifySubscribers(transaction, executionTime);
		} finally {	myTurn.end(); }
	}


	public void addSubscriber(TransactionSubscriber subscriber, long initialTransaction) throws IOException, ClassNotFoundException {
		synchronized (_pendingSubscriptionMonitor) {
			while (_pendingPublications != 0) Thread.yield();
			
			_journal.update(subscriber, initialTransaction);
			super.addSubscriber(subscriber);
		}
	}


	public void close() throws IOException { _journal.close(); }

}

@@ -327,60 +327,8 @@
 g;%0A%0A
-import java.io.IOException;%0Aimport java.util.Date;%0A%0A
 impo
@@ -389,46 +389,8 @@
 on;%0A
-import org.prevayler.foundation.Cool;%0A
 impo
@@ -595,9 +595,77 @@
 hip.
-*
+TransactionCensor;%0A%0Aimport java.io.IOException;%0Aimport java.util.Date
 ;%0A%0Ap
@@ -1124,118 +1124,8 @@
 );%0A%0A
-%09private int _transactionsToJournal = 0;%0A%09private final Object _transactionsToJournalMonitor = new Object();%0A%0A
 %0A%09pu
@@ -2241,23 +2241,31 @@
 urn);%0A%09%09
+_
 journal
+.append
 (transac
@@ -2269,24 +2269,16 @@
 saction,
- myTurn,
  executi
@@ -2279,24 +2279,32 @@
 xecutionTime
+, myTurn
 );%0A%09%09notifyS
@@ -2836,86 +2836,8 @@
 e);%0A
-%09%09%09synchronized (_transactionsToJournalMonitor) %7B _transactionsToJournal++; %7D%0A
 %09%09%09m
@@ -2882,36 +2882,34 @@
 on r) %7B 
-dealWithError(myTurn
+myTurn.alwaysSkip(
 ); throw
@@ -2938,554 +2938,44 @@
 ) %7B%09
-dealWithError(myTurn); throw e; %7D%0A%09%7D%0A%0A%09%0A%09private void dealWithError(Turn myTurn) %7B%0A%09%09synchronized(_transactionsToJournalMonitor) %7B%0A%09%09%09while (_transactionsToJournal != 0) Cool.wait(_transactionsToJournalMonitor);%0A%09%09%7D%0A%09%09myTurn.alwaysSkip();%0A%09%7D%0A%0A%09%0A%09private void journal(Transaction transaction, Turn myTurn, Date executionTime) %7B%0A%09%09_journal.append(transaction, executionTime, myTurn);%0A%09%09synchronized (_transactionsToJournalMonitor) %7B%0A%09%09%09_transactionsToJournal--;%0A%09%09%09if (_transactionsToJournal == 0) _transactionsToJournalMonitor.notify(); %0A%09%09
+myTurn.alwaysSkip(); throw e; 
 %7D%0A%09%7D%0A%0A
-%09
 %0A%09pr

Rollback attempted fix for foodtaster vs. batched journaling bug.