/*
 * 
 * Copyright 2004 Sun Microsystems, Inc. All Rights Reserved.
 * 
 * The contents of this file are subject to the Sun Community Source
 * License v3.0/Jini Technology Specific Attachment v 1.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://wwws.sun.com/software/jini/licensing/SCSL3_JiniTSA1.html.
 * Software distributed under the License is distributed on an "AS IS"
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
 * the License for the specific language governing rights and limitations
 * under the License.
 * 
 * The Reference Code is Jini Technology Core Platform code, v2.0. The
 * Developer of the Reference Code is Sun Microsystems, Inc.
 * 
 * Contributor(s): Sun Microsystems, Inc.
 * 
 * The contents of this file comply with the Jini Technology Core
 * Platform Compatibility Kit, v2.0A.
 * 
 * Tester(s): Sun Microsystems, Inc.
 * 
 * Test Platform(s):
 * Java 2 SDK, Standard Edition, Version 1.4.2_03 for Solaris(TM)SPARC/x86
 * Java 2 SDK, Standard Edition, Version 1.4.2_03 for Linux (Intel x86)
 * Java 2 SDK, Standard Edition, Version 1.4.2_03 for Windows (Intel Platform)
 * 
 */
package com.sun.jini.outrigger.logstore;

import com.sun.jini.outrigger.LogOps;
import com.sun.jini.outrigger.OutriggerServerImpl;
import com.sun.jini.outrigger.StorableObject;
import com.sun.jini.outrigger.StorableResource;
import net.jini.id.Uuid;

import java.io.File;
import java.io.FileDescriptor;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.Observable;
import java.util.logging.Level;
import java.util.logging.Logger;

import net.jini.space.InternalSpaceException;

/**
 * A class to write a log file, to be read later by
 * <code>LogInputFile</code>.  Each operation on the file is forced to
 * disk, so when the operation logging function returns, the data is
 * committed to the log in a recoverable way.
 * <p>
 * <code>LogOutputFile</code> cannot extend <code>Observable</code>
 * because it must extend <code>LogFile</code> (clearly
 * <code>Observable</code> should have been an interface).  It acts as
 * an <code>Observable</code> by having a method that returns its
 * "observable part", which is an object that reports observable
 * events.  Right now the only observable event is the switching to a
 * new physical file when the current one becomes full.
 *
 * @author Sun Microsystems, Inc.
 * @see LogInputFile
 * @see java.util.Observable
 */
class LogOutputFile extends LogFile implements LogOps {
    private RandomAccessFile	logFile = null;// the current output log file
    private FileDescriptor	logFD;	   // the current log file descriptor
    private ObjectOutputStream	out;	   // objects written
    private int			suffix;	   // the current suffix number
    private int			opCnt;	   // number of ops on current file
    private int			maxOps;	   // max ops to allow in file
    private Observable		observable;// handle Observer/Observable

    private long logBytes = 0;
    private final byte[] intBuf = new byte[4];
    private final byte[] zeroBuf = new byte[4];

    private long deferedUpdateLength = 0;
    private long deferedPosition = 0;

    private static final long intBytes = 4;

    /** Logger for logging persistent store related information */
    private static final Logger logger = 
	Logger.getLogger(OutriggerServerImpl.storeLoggerName);

    /**
     * Create a <code>LogOutputFile</code> object that will stream
     * output to a series of files described by <code>basePath</code>,
     * as interpreted by the relevant <code>LogFile</code>
     * constructor.  When the file becomes full (the maximum number of
     * operations is reached), the file is closed and a new file with
     * the next highest suffix is created.  The
     * <code>Observable</code> notification for this event passes a
     * <code>File</code> argument for the filled file as the argument
     * to <code>Observer</code>.
     *
     * @see LogStream#LogStream(String)
     * @see observable() 
     */
    LogOutputFile(String basePath, int maxOps) throws IOException {
	super(basePath);
	ArrayList inDir = new ArrayList();
	suffix = existingLogs(inDir);
	this.maxOps = maxOps;
	nextPath();
    }

    /**
     * Return an <code>Observable</code> object that represents this object
     * in the Observer/Observable pattern.
     *
     * @see java.util.Observer
     */
    Observable observable() {
	if (observable == null) {	     // defer allocation until needed
	    observable = new Observable() {  // we only use this if changed
		public void notifyObservers() {
		    setChanged();
		    super.notifyObservers();
		}
		public void notifyObservers(Object arg) {
		    setChanged();
		    super.notifyObservers(arg);
		}
	    };
	}
	return observable;
    }

    /**
     * Switch this over to the next path in the list
     */
    private void nextPath() throws IOException {
	boolean completed = false;

	if (logFile != null) {

	    // If there was a deferred header, write it out now
	    //
	    if (deferedUpdateLength != 0) {
		logFile.seek(deferedPosition);
		writeInt((int)deferedUpdateLength);
	    }
	    logFD.sync();		// force the bytes to disk
	    try {
		close();   	        // close the stream and the file
	    } catch (IOException ignore) { } // assume this is okay
	    completed = true;
	}

	suffix++;			// go to next suffix
	logFile = new RandomAccessFile(baseDir.getPath() + File.separator +
				       baseFile + suffix, "rw");
	logFD = logFile.getFD();
	out = new ObjectOutputStream(new LogOutputStream(logFile));

	writeInt((int)LOG_VERSION);

	logBytes = logFile.getFilePointer();
	logFile.setLength(logBytes);

	// always start out with zero length header for the next update
	logFile.write(zeroBuf);

	// force length header to disk 
	logFD.sync();

	deferedUpdateLength = 0;
	opCnt = 0;

	/*
	 * Tell consumer about the completed log.  This is done after the
	 * new one is created so that the old path can be known not
	 * to be the newest (because something newer is there).
	 */
	if (observable != null && completed)
	    observable.notifyObservers();
    }

    /**
     * Close the log, but don't remove it.
     */
    synchronized void close() throws IOException {
	if (logFile != null) {
	    try {
		out.close();
		logFile.close();
	    } finally {
		logFile = null;
	    }
	}
    }

    /**
     * Override destroy so we can try to close logFile before calling
     * super tries to delete all the files.
     */
    void destroy() {
	try {
	    close();
	} catch (Throwable t) {
	    // Don't let failure keep us from deleting the files we can	    
	}
	super.destroy();
    }

    /**
     * Log a server boot.
     */
    public synchronized void bootOp(long time, long sessionId) {
	try {
	    out.writeByte(BOOT_OP);
	    out.writeLong(time);
	    out.writeLong(sessionId);
	    flush();
	} catch (IOException e) {
	    failed(e);
	}
    }

    /**
     * Log a change in join state
     */
    public synchronized void joinStateOp(StorableObject state) {
	try {
	    out.writeByte(JOINSTATE_OP);
	    out.writeObject(new BaseObject(state));
	    flush();
	} catch (IOException e) {
	    failed(e);
	}
    }

    /**
     * Log a <code>write</code> operation.
     */
    public synchronized void writeOp(StorableResource entry, Long txnId) {
	try {
	    out.writeByte(WRITE_OP);
	    out.writeObject(new Resource(entry));
	    out.writeObject(txnId);

	    // A write operation under a transaction does not need to be
	    // flushed until it is prepared.
	    //
	    flush(txnId == null);
	} catch (IOException e) {
	    failed(e);
	}
    }

    /**
     * Log a <code>take</code> operation.
     */
    public synchronized void takeOp(Uuid cookie, Long txnId) {
	try {
	    out.writeByte(TAKE_OP);
	    cookie.write(out);
	    out.writeObject(txnId);

	    // A take operation under a transaction does not need to be
	    // flushed until it is prepared.
	    //
	    flush(txnId == null);
	} catch (IOException e) {
	    failed(e);
	}
    }

    /**
     * Log a <code>notify</code> operation.
     */
    public synchronized void registerOp(StorableResource registration,
					StorableObject template) {
	try {
	    out.writeByte(REGISTER_OP);
	    out.writeObject(new Registration(registration, template));
	    flush();
	} catch (IOException e) {
	    failed(e);
	}
    }

    /**
     * Log a <code>renew</code> operation.
     */
    public synchronized void renewOp(Uuid cookie, long expiration) {
	try {
	    out.writeByte(RENEW_OP);
	    cookie.write(out);
	    out.writeLong(expiration);
	    flush();
	} catch (IOException e) {
	    failed(e);
	}
    }

    /**
     * Log a <code>cancel</code> operation.
     */
    public synchronized void cancelOp(Uuid cookie, boolean expired) {
	try {
	    out.writeByte(CANCEL_OP);
	    cookie.write(out);

	    // cancels due to expiration don't need to be flushed
	    // right away
	    flush(!expired);
	} catch (IOException e) {
	    failed(e);
	}
    }

    /**
     * Log a transaction <code>prepare</code> operation.
     */
    public synchronized void prepareOp(Long txnId,
				       StorableObject transaction) {
	try {
	    out.writeByte(PREPARE_OP);
	    out.writeObject(txnId);
	    out.writeObject(new BaseObject(transaction));
	    flush();
	} catch (IOException e) {
	    failed(e);
	}
    }

    /**
     * Log a transaction <code>commit</code> operation.
     */
    public synchronized void commitOp(Long txnId) {
	try {
	    out.writeByte(COMMIT_OP);
	    out.writeObject(txnId);
	    flush();
	} catch (IOException e) {
	    failed(e);
	}
    }

    /**
     * Log a transaction <code>abort</code> operation.
     */
    public synchronized void abortOp(Long txnId) {
	try {
	    out.writeByte(ABORT_OP);
	    out.writeObject(txnId);
	    flush();
	} catch (IOException e) {
	    failed(e);
	}
    }

    public synchronized void uuidOp(Uuid uuid) {
	try {
	    out.writeByte(UUID_OP);
	    uuid.write(out);
	    flush();
	} catch (IOException e) {
	    failed(e);
	}
    }

    /**
     * Flush the current output after an operation.  If the number of
     * operations is exceeded, shift over to the next path.  
     */
    private void flush() throws IOException {
	flush(true);
    }

    /**
     * Conditionally flush the current output. If the number of
     * operations is exceeded, shift over to the next path even if
     * <code>forceToDisk</code> is <code>false</code>.
     */
    private synchronized void flush(boolean forceToDisk) throws IOException {

	out.flush();

	if (forceToDisk) {

	    // must force contents to disk before writing real length header
	    logFD.sync();
	}

	long entryEnd = logFile.getFilePointer();
	long updateLen = entryEnd - logBytes - intBytes;

        // If we are not forcing to disk, we want to defer the write of the
        // first header. This will leave a zero just after the last sync'ed
        // record and will assure that LogInputFile will not read a partially
        // written record.
        //
        if (!forceToDisk) {

	    // If this is the first flush(false) we save the header information
	    // and location for later. Otherwise we write out the header
	    // normally.
	    //
	    if (deferedUpdateLength == 0) {
		deferedUpdateLength = updateLen;  // save the header length
		deferedPosition = logBytes;       // and position for later
	    } else {
		// write real length header
		logFile.seek(logBytes);
		writeInt((int)updateLen);
	    }
	} else {

	    // If there was a deferred header, write that out now and
	    // then write the current header.
	    //
	    if (deferedUpdateLength != 0) {
		logFile.seek(deferedPosition);
		writeInt((int)deferedUpdateLength);
		deferedUpdateLength = 0;
	    }
	    // write real length header
	    logFile.seek(logBytes);
	    writeInt((int)updateLen);
	}

	// pad out update record so length header does not span disk blocks
	entryEnd = (entryEnd + 3) & ~3L;

	// write zero length header for next update
	logFile.seek(entryEnd);
	logFile.write(zeroBuf);
	logBytes = entryEnd;

	if (forceToDisk)
	    logFD.sync();

	if (++opCnt >= maxOps)
	    nextPath();
	else
	    out.reset();		// not critical to flush this
    }

    /**
     * Write an int value in single write operation.
     *   
     * @param val int value
     * @throws IOException if any other I/O error occurs
     */  
    private void writeInt(int val) throws IOException {
	intBuf[0] = (byte) (val >> 24);
	intBuf[1] = (byte) (val >> 16);
	intBuf[2] = (byte) (val >> 8);
	intBuf[3] = (byte) val;
	logFile.write(intBuf);
    }

    private void failed(Exception e) throws InternalSpaceException {
	logger.log(Level.SEVERE, 
		   "Unexpected I/O error while persisting Space data",
		   e);
	System.exit(-5);
    }
}
