/*
* Copyright (c) 2009 Nokia Corporation and/or its subsidiary(-ies).
* All rights reserved.
* This component and the accompanying materials are made available
* under the terms of the License "Eclipse Public License v1.0"
* which accompanies this distribution, and is available
* at the URL "http://www.eclipse.org/legal/epl-v10.html".
*
* Initial Contributors:
* Nokia Corporation - initial contribution.
*
* Contributors:
*
* Description: 
*
*/
/**
 * 
 */
package com.nokia.tcf.impl;

import java.io.IOException;

import org.eclipse.core.runtime.IStatus;

import com.nokia.tcf.api.ITCMessage;
import com.nokia.tcf.api.ITCMessageInputStream;
import com.nokia.tcf.api.TCErrorConstants;
import com.nokia.tcf.api.TCFClassFactory;

public class TCMessageInputStream implements ITCMessageInputStream {

	private long inputStreamSize;
	private long clientId;
	private TCAPIConnection api;
	private boolean isOpen = false;
	private int blockingTime = 10;
	private long MAX_BYTES = 2*1024*1024;

	/**
	 * Constructor
	 * 
	 * @param inputStreamSize
	 * @param overFlowToFile
	 * @param clientId
	 */
	public TCMessageInputStream(TCAPIConnection apiConnection, long inputStreamSize, long clientId) {
		this.inputStreamSize = inputStreamSize;
		this.clientId = clientId;
		api = apiConnection;
	}

	public int peekMessages() throws IOException {
		int number = 0;
		if (!this.isOpen) {
			// stream not open
			String errString = TCErrorConstants.getErrorMessage(TCErrorConstants.TCAPI_ERR_INPUTSTREAM_CLOSED);
			IOException e = new IOException(errString);
			e.fillInStackTrace();
			throw e;
		}
		// poll for # messages in stream
		long[] outNumberTotalMessages = new long[1];
		long ret = api.nativePollInputStream(this.clientId, outNumberTotalMessages);
		if (ret != TCErrorConstants.TCAPI_ERR_NONE) {
			// error on first poll
			String errString = TCErrorConstants.getErrorMessage((int)ret);
			IOException e = new IOException(errString);
			e.fillInStackTrace();
			throw e;
		}
		number = (int)outNumberTotalMessages[0];
		
		return number;
	}

	public boolean isOpen() {
		return this.isOpen;
	}

	public void open() throws IOException {
		if (!this.isOpen) {
			// Remove overflow to file and input stream file path arguments from nativeOpenInputStream()
			long ret = api.nativeOpenInputStream(clientId, null, inputStreamSize, false);
			if (ret == TCErrorConstants.TCAPI_ERR_NONE) {
				isOpen = true;
			} else {
				String errString = TCErrorConstants.getErrorMessage((int)ret);
				IOException e = new IOException(errString);
				e.fillInStackTrace();
				throw e;
			}
		}
	}

	public ITCMessage readMessage() throws IOException {
		ITCMessage message = null;

		// if not open throw exception
		if (!this.isOpen) {
			// stream not open
			String errString = TCErrorConstants.getErrorMessage(TCErrorConstants.TCAPI_ERR_INPUTSTREAM_CLOSED);
			IOException e = new IOException(errString);
			e.fillInStackTrace();
			throw e;
		}

		// poll for # messages in stream
		long[] outNumberTotalMessages = new long[1];
		outNumberTotalMessages[0] = 0;
		do {
			if (!this.isOpen) {
				// stream not open
				String errString = TCErrorConstants.getErrorMessage(TCErrorConstants.TCAPI_ERR_INPUTSTREAM_CLOSED);
				IOException e = new IOException(errString);
				e.fillInStackTrace();
				throw e;
			}
			long ret = api.nativePollInputStream(this.clientId, outNumberTotalMessages);
			if (ret != TCErrorConstants.TCAPI_ERR_NONE) {
				// error on first poll
				String errString = TCErrorConstants.getErrorMessage((int)ret);
				IOException e = new IOException(errString);
				e.fillInStackTrace();
				throw e;
			}
			if (outNumberTotalMessages[0] == 0) {
				try {
					Thread.sleep(blockingTime);
				} catch (InterruptedException e) {
					break;
				}
			}
		} while (outNumberTotalMessages[0] == 0);
		
		// check # in stream
		if (outNumberTotalMessages[0] <= 0) {
			// no messages in stream
			return message;
		}
		// get number of bytes in first message
		long[] outNumberBytesInMessage = new long[1];
		long ret = api.nativeGetInputStreamMessageBytes(this.clientId, 1, outNumberBytesInMessage);
		if (ret != TCErrorConstants.TCAPI_ERR_NONE) {
			// error on getting bytes in message
			String errString = TCErrorConstants.getErrorMessage((int)ret);
			IOException e = new IOException(errString);
			e.fillInStackTrace();
			throw e;
		}

		// OK, read first message
		long totalBytes = outNumberBytesInMessage[0];
		byte[] outMessage = new byte[(int)totalBytes];
		long[] outNumberMessages = new long[1];
		long[] outNumberBytesRead = new long[1];
		ret = api.nativeReadInputStream(this.clientId, 1, outNumberMessages, outNumberBytesRead, totalBytes, outMessage);
		if (ret != TCErrorConstants.TCAPI_ERR_NONE) {
			// error on read message
			String errString = TCErrorConstants.getErrorMessage((int)ret);
			IOException e = new IOException(errString);
			e.fillInStackTrace();
			throw e;
		}
		
		// OK, pass back the message
		message = TCFClassFactory.createITCMessage(outMessage);
		
		return message;
	}

	public ITCMessage[] readMessages(int inNumberMessages) throws IOException {
		ITCMessage[] messages = null;

		// if not open throw exception
		if (!this.isOpen) {
			// stream not open
			String errString = TCErrorConstants.getErrorMessage(TCErrorConstants.TCAPI_ERR_INPUTSTREAM_CLOSED);
			IOException e = new IOException(errString);
			e.fillInStackTrace();
			throw e;
		}
		// poll for # messages in stream
		long[] outNumberTotalMessages = new long[1];
		outNumberTotalMessages[0] = 0;
		do {
			if (!this.isOpen) {
				// stream not open
				String errString = TCErrorConstants.getErrorMessage(TCErrorConstants.TCAPI_ERR_INPUTSTREAM_CLOSED);
				IOException e = new IOException(errString);
				e.fillInStackTrace();
				throw e;
			}
			long ret = api.nativePollInputStream(this.clientId, outNumberTotalMessages);
			if (ret != TCErrorConstants.TCAPI_ERR_NONE) {
				// error on first poll
				String errString = TCErrorConstants.getErrorMessage((int)ret);
				IOException e = new IOException(errString);
				e.fillInStackTrace();
				throw e;
			}
			if (outNumberTotalMessages[0] == 0) {
				try {
					Thread.sleep(blockingTime);
				} catch (InterruptedException e) {
					break;
				}
			}
		} while(outNumberTotalMessages[0] == 0);
		
		// check # in stream
		if (outNumberTotalMessages[0] <= 0) {
			// no messages in stream
			return messages;
		}
		if (inNumberMessages == 0) {
			inNumberMessages = (int)outNumberTotalMessages[0];
		} else if (inNumberMessages > outNumberTotalMessages[0]) {
			inNumberMessages = (int)outNumberTotalMessages[0];
		}
		// get number of bytes in each message
		long[] outNumberBytesInMessages = new long[inNumberMessages];
		long ret = api.nativeGetInputStreamMessageBytes(this.clientId, inNumberMessages, outNumberBytesInMessages);
		if (ret != TCErrorConstants.TCAPI_ERR_NONE) {
			// error on getting bytes in message
			String errString = TCErrorConstants.getErrorMessage((int)ret);
			IOException e = new IOException(errString);
			e.fillInStackTrace();
			throw e;
		}
		// read all requested messages
		int totalBytes = 0;
		for (int i = 0; i < inNumberMessages; i++) {
			totalBytes += outNumberBytesInMessages[i];
		}
		if (totalBytes > MAX_BYTES) totalBytes = (int)MAX_BYTES;
		byte[] messageData = new byte[totalBytes];
		long[] outNumberMessages = new long[1];
		long[] outNumberBytesRead = new long[1];
		ret = api.nativeReadInputStream(this.clientId, inNumberMessages, outNumberMessages, outNumberBytesRead, totalBytes, messageData);
		if (ret != TCErrorConstants.TCAPI_ERR_NONE) {
			// error on getting bytes in message
			String errString = TCErrorConstants.getErrorMessage((int)ret);
			IOException e = new IOException(errString);
			e.fillInStackTrace();
			throw e;
		}
		int messageDataStart = 0;
		messages = new TCMessage[(int)outNumberMessages[0]];
		for (int i = 0; i < outNumberMessages[0]; i++) {
			int size = (int)outNumberBytesInMessages[i];
			byte[] data = new byte[size];
			for (int j = 0; j < size; j++) {
				data[j] = messageData[messageDataStart + j];
			}
			messages[i] = TCFClassFactory.createITCMessage(data);
			messageDataStart += size;
		}
		return messages;
	}

	public void close() throws IOException {
		if (this.isOpen) {
			// Stop capturing first
			IStatus status = api.stop();
			if (status.isOK()) {
				// ok, now close the stream
				long ret = api.nativeCloseInputStream(this.clientId);
				if (ret == TCErrorConstants.TCAPI_ERR_NONE) {
					isOpen = false;
				} else {
					String errString = TCErrorConstants.getErrorMessage((int)ret);
					IOException e = new IOException(errString);
					e.fillInStackTrace();
					throw e;
				}
			} else {
				String errString = status.getMessage();
				IOException e = new IOException(errString);
				e.fillInStackTrace();
				throw e;
			}
		}		
	}
	public byte[] readBytes(int inNumberMessages) throws IOException {
		// if not open throw exception
		if (!this.isOpen) {
			// stream not open
			String errString = TCErrorConstants.getErrorMessage(TCErrorConstants.TCAPI_ERR_INPUTSTREAM_CLOSED);
			IOException e = new IOException(errString);
			e.fillInStackTrace();
			throw e;
		}
//		inNumberMessages = 0; // 0 --> get all that fit in MAX_BYTES
		byte[] messageData = new byte[(int) MAX_BYTES];
		long[] outNumberMessagesRead = new long[1];
		outNumberMessagesRead[0] = 0;
		long[] outNumberBytesRead = new long[1];
		do {
			long ret = api.nativeReadInputStream(this.clientId, inNumberMessages, outNumberMessagesRead, outNumberBytesRead, MAX_BYTES, messageData);
			if (ret != TCErrorConstants.TCAPI_ERR_NONE) {
				// error on getting bytes in message
				String errString = TCErrorConstants.getErrorMessage((int)ret);
				IOException e = new IOException(errString);
				e.fillInStackTrace();
				throw e;
			}
			if (outNumberMessagesRead[0] == 0) {
				try {
					Thread.sleep(blockingTime);
				} catch (InterruptedException e) {
					break;
				}
			}
		} while (outNumberMessagesRead[0] == 0);

		// return how many messages actually processed
		byte[] newMessageData = new byte[(int)outNumberBytesRead[0]];
		for (int i = 0; i < outNumberBytesRead[0]; i++) {
			newMessageData[i] = messageData[i]; 
		}
		return newMessageData;
	}
	public byte[] readBytes(int inNumberMessages, int[] outNumberMessages) throws IOException {
		outNumberMessages[0] = 0;
		
		// if not open throw exception
		if (!this.isOpen) {
			// stream not open
			String errString = TCErrorConstants.getErrorMessage(TCErrorConstants.TCAPI_ERR_INPUTSTREAM_CLOSED);
			IOException e = new IOException(errString);
			e.fillInStackTrace();
			throw e;
		}
		inNumberMessages = 0; // 0 --> get all that fit in MAX_BYTES
		byte[] messageData = new byte[(int) MAX_BYTES];
		long[] outNumberMessagesRead = new long[1];
		outNumberMessagesRead[0] = 0;
		long[] outNumberBytesRead = new long[1];
		do {
			long ret = api.nativeReadInputStream(this.clientId, inNumberMessages, outNumberMessagesRead, outNumberBytesRead, MAX_BYTES, messageData);
			if (ret != TCErrorConstants.TCAPI_ERR_NONE) {
				// error on getting bytes in message
				String errString = TCErrorConstants.getErrorMessage((int)ret);
				IOException e = new IOException(errString);
				e.fillInStackTrace();
				throw e;
			}
			if (outNumberMessagesRead[0] == 0) {
				try {
					Thread.sleep(blockingTime);
				} catch (InterruptedException e) {
					break;
				}
			}
		} while (outNumberMessagesRead[0] == 0);
		
		// return how many messages actually processed
		byte[] newMessageData = new byte[(int)outNumberBytesRead[0]];
		for (int i = 0; i < outNumberBytesRead[0]; i++) {
			newMessageData[i] = messageData[i]; 
		}
		outNumberMessages[0] = (int)outNumberMessagesRead[0];
		return newMessageData;
		
	}
}