/*
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 *
 * Copyright (c) 2000-2010 Oracle and/or its affiliates. All rights reserved.
 *
 * The contents of this file are subject to the terms of either the GNU
 * General Public License Version 2 only ("GPL") or the Common Development
 * and Distribution License("CDDL") (collectively, the "License").  You
 * may not use this file except in compliance with the License.  You can
 * obtain a copy of the License at
 * https://glassfish.dev.java.net/public/CDDL+GPL_1_1.html
 * or packager/legal/LICENSE.txt.  See the License for the specific
 * language governing permissions and limitations under the License.
 *
 * When distributing the software, include this License Header Notice in each
 * file and include the License file at packager/legal/LICENSE.txt.
 *
 * GPL Classpath Exception:
 * Oracle designates this particular file as subject to the "Classpath"
 * exception as provided by Oracle in the GPL Version 2 section of the License
 * file that accompanied this code.
 *
 * Modifications:
 * If applicable, add the following below the License Header, with the fields
 * enclosed by brackets [] replaced by your own identifying information:
 * "Portions Copyright [year] [name of copyright owner]"
 *
 * Contributor(s):
 * If you wish your version of this file to be governed by only the CDDL or
 * only the GPL Version 2, indicate your decision by adding "[Contributor]
 * elects to include this software in this distribution under the [CDDL or GPL
 * Version 2] license."  If you don't indicate a single choice of license, a
 * recipient has the option to distribute your version of this file under
 * either the CDDL, the GPL Version 2 or to extend the choice of license to
 * its licensees as provided above.  However, if you add GPL Version 2 code
 * and therefore, elected the GPL Version 2 license, then the option applies
 * only if the new code is made subject to such option by the copyright
 * holder.
 */

/*
 * @(#)FileLogRecordIterator.java	1.2 06/29/07
 */ 

package com.sun.messaging.jmq.util.txnlog.file;


import java.util.Date;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;

import com.sun.messaging.jmq.util.txnlog.*;

/**
 * This class reads MQ transaction log file from the last check point to the end
 * of the file. 
 */

public class FileLogRecordIterator implements Iterator {
	
	//private boolean hasNext = false;
	private FileTransactionLogWriter lwriter = null;
	private RandomAccessFile raf = null;
	private long cpPosition = -1;
        
    private long cpSequence = -1;
	
	private FileTransactionLogRecord entry = null;
	
	private FileTransactionLogRecord[] compoundEntries = null;

	private int compoundEntryIndex =0;
	
	private boolean eof = false;
	
	private long ct = 0;
    
    private long entrySequence = 0;
	
	private boolean debug = false;
	
	public FileLogRecordIterator (FileTransactionLogWriter writer) throws IOException {
		
        this.debug = Boolean.getBoolean (FileTransactionLogWriter.TXNLOG_DEBUG_PROP_NAME);
        
        this.lwriter = writer;
		this.raf = lwriter.getRAF();
		this.cpPosition = lwriter.getCPPosition();
		this.cpSequence = lwriter.getCPSequence();
                
		raf.seek(cpPosition);
		
		readNextRecord();
	}
	
	/**
	 * 
	 */
	public synchronized boolean hasNext() {

		/**
		 * We always read one record ahead.
		 */
		if (entry != null) {
			return true;
		} else {
			return false;
		}

	}
	
	public synchronized Object next() {
		
		if (eof) {
			throw new NoSuchElementException ();
		}
		
		if ( entry != null ) {
			
			//save the current entry to be returned.
			TransactionLogRecord lrec = entry;
			
			if(compoundEntries!=null)
			{
				// entry is a member of a compound log entry
				
				if(compoundEntryIndex < (compoundEntries.length-1) )
				{				
					// there are more compound entries to go
					compoundEntryIndex++;
					entry = compoundEntries[compoundEntryIndex];
					if(debug)
						log("setting next entry to a compound sub record");
					
					// no need to read next record
					return lrec;
				}
				else
				{
					// this is the last compound entry
					if(debug)
						log("This is the last compound sub record");
					compoundEntries = null;
					compoundEntryIndex = 0;
				}				
				
				
			}
			
			//set to null so that the next read can assign to it if exists.
			entry = null;
			
			//read the next record.
			readNextRecord();
			
			//return the entry
			return lrec;
		} 
		
		throw new NoSuchElementException ();
	}
	
	public void remove() {
		throw new UnsupportedOperationException ("Unsupported Operation.");
	}
	
	private synchronized void readNextRecord() {

		if(debug)
			log("readNextRecord");
		if (eof) {
			return;
		}

		try {

			byte[] bytes = new byte[FileTransactionLogWriter.RECORD_HEADER_SIZE];

			//read record header
			int bread = raf.read(bytes);

			//if bytes read is less than header size, we reached eof
			if (bread < FileTransactionLogWriter.RECORD_HEADER_SIZE) {

				log("Reached end of file., records read: " + ct);
				eof = true;

				return;
			}

			//read header fields
			ByteBuffer buf = ByteBuffer.wrap(bytes);

			// 1. magic number
			int magic = buf.getInt();

			// 2. record type
			int rtype = buf.getInt();

			// 3. body size
			int bsize = buf.getInt();

			// 4. timestamp
			long timestamp = buf.getLong();

			//check magic number
			if (magic != FileTransactionLogWriter.RECORD_MAGIC_NUMBER) {
                            
				log("Reached end of file. No Magic number., records read: " +  ct);
				
				eof = true;
				
				return;
			}
			
			//5. entry seq number
			long seq = buf.getLong();
			
			//6. cp seq number
			long cpseq = buf.getLong();

			// read record body
			byte[] body = new byte[bsize];
			int rsize = raf.read(body);

			if (rsize < bsize) {
				
				log("Reached end of file. Body read reached EOF. records read: " +  ct);
                                
				eof = true;
				
				return;
			}

			// 7. check sum
			long valueSet = buf.getLong();
			long calculated = lwriter.calculateCheckSum(body);

			if (valueSet != calculated) {
                
                log ("Reached end of file.  Check sum not validate., records read: " +  ct);
                
                eof = true;
                return;
            }
            
            if (cpseq != this.cpSequence) {
                
                log ("Reached end of check point.  records read: " + ct);
                
                eof = true;
                
                return;
            }
            
            //validate entry sequence
            if (seq != this.entrySequence) {
                log ("Entry sequence is not valid. Expected: " + entrySequence + ", but read: " + seq + ", total records read: " + ct);
                
                eof = true;
                
                return;
            } 
            
            entrySequence ++;
            
            //construct next available log record object
            entry = new FileTransactionLogRecord (timestamp, rtype, seq);
            
            entry.setCheckPointSequence (cpseq);
            
            entry.setBody (body);
            
            if(entry.getType()== TransactionLogType.COMPOUND_TRANSACTION)
            {
            	processCompoundTransactionLogRecord();
            }
            
            ct ++;
			
		} catch (Exception e) {
			eof = true;
		}

		return;
	}
	
	private void processCompoundTransactionLogRecord()
	{
		TransactionLogRecord compoundRecord =  entry;
		byte[] compoundBody = compoundRecord.getBody();
		
		//read header fields
		ByteBuffer buf = ByteBuffer.wrap(compoundBody);
		
		int numEntries  = buf.getInt();
		if(debug)
			log("processCompoundTransactionLogRecord numEntries="+numEntries);
		
		compoundEntries = new FileTransactionLogRecord[numEntries];
		compoundEntryIndex = 0;
		for(int i=0;i<numEntries;i++)
		{
			
			FileTransactionLogRecord subEntry = new FileTransactionLogRecord (entry.getTimestamp(), entry.getType(), entry.getSequence());
			
			subEntry.setCheckPointSequence(entry.getCheckPointSequence());
			
			compoundEntries[i]=subEntry;
			
			// 2. record type
			int rtype = buf.getInt();
			subEntry.setType(rtype);

			// 3. body size
			int bsize = buf.getInt();
			
			byte[] subBody = new byte[bsize];
			buf.get(subBody);			
			
			subEntry.setBody(subBody);
		}
		
		//set entry to the first record
		entry = compoundEntries[0];
		
	}
	
	private void log (String msg) {
    	if ( debug ) {
    		System.out.println(new Date() + ": " + msg);
    	}
    }

	/**
	 * @param args
	 */
	public static void main (String[] args) throws Exception {
        
        String fname = "MQTxn.log";
        
        if (args.length > 0) {
            fname = args[0];
        }
        
        int nrec = 0;
        
        TransactionLogWriter lwriter = new FileTransactionLogWriter (fname);
        
        if (lwriter.playBackRequired ()) {
            
            Iterator it = lwriter.iterator ();
            nrec = 0;
            while (it.hasNext ()) {
                Object obj = it.next ();
                System.out.println (obj.toString ());
                nrec ++;
            }
        } else {
            System.out.println ("*** No playback is required.");
        }
        
        System.out.println ("*** Read Txn log file: " + fname + ", nrec: " + nrec);
    }

}
