/*
       This file is part of quExec.
   
       quExec is free software; you can redistribute it and/or modify
       it under the terms of the GNU Lesser General Public License as published by
       the Free Software Foundation; either version 2 of the License, or
       (at your option) any later version.
   
       quExec 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 Lesser General Public License for more details.
  
      You should have received a copy of the GNU Lesser General Public License
      along with quExec; if not, write to the Free Software
      Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
  */
  
  package net.sourceforge.quexec.packet.chars.stream;
  
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertFalse;
  import static org.junit.Assert.assertNull;
  import static org.junit.Assert.assertTrue;
  
  import java.util.concurrent.TimeoutException;
  
  import net.sourceforge.quexec.testutil.AbstractMultithreadedTest;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.junit.Ignore;
  import org.junit.Test;
  
  /**
   * @author schickin
   *
   */
  public abstract class AbstractCharPacketStreamTest
  extends AbstractMultithreadedTest {
  	
  	private static final Log log = LogFactory.getLog(AbstractCharPacketStreamTest.class);
  	
  	private static final String msg = "Hello!";
  	
  	private CharPacketInputStream inputStream;
  	
  	private CharPacketOutputStream outputStream;
  
  	protected final void initStreamsBeforeEveryTest() throws Exception {
  		this.inputStream = getInputStream();
  		this.outputStream = getOutputStream();
  	}
  
  	protected abstract CharPacketInputStream getInputStream();
  	
  	protected abstract CharPacketOutputStream getOutputStream();
  
  	@Test
  	public void emptyTransmission()
  	throws InterruptedException, TimeoutException {
  		sequentialSendReadCheck(0);
  	}
  	
  	@Test
  	public void singlePacketTransmission()
  	throws InterruptedException, TimeoutException {
  		sequentialSendReadCheck(1);
  	}
  	
  	@Test
  	public void severalPacketsTransmission() throws
  	InterruptedException, TimeoutException {
  		sequentialSendReadCheck(10);
  	}
  	
  	// TODO: This test reports transient errors (e.g. Hello!8 expected but gut Hello!18)
  	// Analysis showed so far that following things happen:
  	//
  	// 1. the underlying ActiveMQ connection fails with an exception:
  	//
  	//	5059 ERROR [net.sourceforge.quexec.packet.chars.stream.JmsCharPacketStreamTest] JMSException recorded
  	//	javax.jms.JMSException: java.io.EOFException
  	//	at net.sourceforge.quexec.packet.chars.stream.JmsCharPacketStreamTest$1.onException(JmsCharPacketStreamTest.java:69)
  	//	at org.springframework.jms.connection.ChainedExceptionListener.onException(ChainedExceptionListener.java:60)
  	//	at org.apache.activemq.ActiveMQConnection$4.run(ActiveMQConnection.java:1779)
  	//	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1110)
  	//	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603)
  	//	at java.lang.Thread.run(Thread.java:636)
  	//	Caused by: java.io.EOFException
  	//	at java.io.DataInputStream.readInt(DataInputStream.java:392)
  	//	at org.apache.activemq.openwire.OpenWireFormat.unmarshal(OpenWireFormat.java:269)
  	//	at org.apache.activemq.transport.tcp.TcpTransport.readCommand(TcpTransport.java:210)
  	//	at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:202)
  	//	at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:185)
  	//	... 1 more
  	//
  	// 2. the Spring SingleConnectionFactory automatically reconnects
  	//
 	// 3. the recovery mechanism of ActiveMQ reports some duplicate messages in its
 	// message store. seemingly, the recovery does not work correctly.
 	//
 	// Summary: the behavior seems to be caused by non-thread-safe logic within
 	// the embedded ActiveMQ broker that is used for the unit tests
 	
 	@Ignore
 	@Test(timeout = 20000)
 	public void parallelTransmission() throws Exception {
 		log.debug("Test parallelTransmission begins for class " + getClass().getName());
 		parallelSendReadCheck(50);
 	}
 	
 	private void sequentialSendReadCheck(int numMsg)
 	throws InterruptedException, TimeoutException {
 		assertFalse(this.inputStream.isFinished());
 		assertNull(this.inputStream.tryReadPacket());
 
 		for (int i = 0 ; i < numMsg; i++) {
 			this.outputStream.sendPacket(msg + i);
 		}
 		this.outputStream.finish();
 		
 		for (int i = 0 ; i < numMsg; i++) {
 			assertFalse(this.inputStream.isFinished());
 			assertEquals(msg + i, this.inputStream.readPacket());
 		}
 		
 		int retry;
 		for (retry = 0; retry < 100; retry++) {
 			// since the call to isFinished is non-blocking, it may happen that
 			// the finish-packet has not yet reached the inputStream object.
 			// thus, we need to retry a couple of times.
 			if (this.inputStream.isFinished()) {
 				break;
 			}
 			Thread.sleep(10);
 		}
 		assertFalse(retry == 100);
 	}
 	
 	private void parallelSendReadCheck(final int numMsg)
 	throws InterruptedException {
 		Task sender = new Task() {
 			public void doIt() {
 				for (int i = 0 ; i < numMsg; i++) {
 					outputStream.sendPacket(msg + i);
 				}
 				outputStream.finish();
 			};
 		};
 		
 		Task receiver = new Task() {
 			public void doIt() throws Throwable {
 				for (int i = 0 ; i < numMsg; i++) {
 					assertFalse(inputStream.isFinished());
 					assertEquals(msg + i, inputStream.readPacket());
 				}
 				while (!inputStream.isFinished()) {
 					Thread.sleep(10);
 				}
 				assertTrue(inputStream.isFinished());
 			}
 		};
 
 		scheduleTask(sender, "Sender");
 		scheduleTask(receiver, "Receiver");
 		
 		runAll();
 	}
 }