src/test/java/org/apache/ode/jacob/oo/SequentialProcessingTest.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.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://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.ode.jacob.oo;


import org.apache.ode.jacob.Jacob;
import org.apache.ode.jacob.RunnableProcess;
import org.apache.ode.jacob.oo.ProcessUtil.Synchronized;
import org.apache.ode.jacob.vpu.ExecutionQueueImpl;
import org.apache.ode.jacob.vpu.JacobVPU;
import org.junit.Assert;
import org.junit.Test;

import static org.apache.ode.jacob.Jacob.instance;
import static org.apache.ode.jacob.Jacob.object;
import static org.apache.ode.jacob.oo.ProcessUtil.intercept;
import static org.apache.ode.jacob.oo.ProcessUtil.receive;
import static org.apache.ode.jacob.oo.ProcessUtil.sequence;
import static org.apache.ode.jacob.oo.ProcessUtil.sync;


public class SequentialProcessingTest {

    @SuppressWarnings("serial")
	@Test
    public void testParallelProcesses() {
    	final StringBuffer out = new StringBuffer();
    	executeProcess(new RunnableProcess() {
			public void run() {
				instance(intercept(sync(null, atomicProcess("A", out)), atomicProcess("0", out)));
				instance(intercept(sync(null, atomicProcess("B", out)), atomicProcess("1", out)));
			}
        });
    	// parallelism is proven by process "B" being executed before "A0" and "A1"
    	Assert.assertEquals("AB01", out.toString());
    }

    @SuppressWarnings("serial")
	@Test
    public void testSynchronizeProcess() {
    	final StringBuffer out = new StringBuffer();
    	executeProcess(new RunnableProcess() {
			public void run() {
		        Synch c1 = Jacob.newChannel(Synch.class, "");
		        object(receive(c1, new Synch() {
					public void ret() {
						out.append("x");
					}
				}));
				instance(sync(c1, atomicProcess("A", out)));
			}
        });
    	// Return hook "x" is executed after process "A" 
    	Assert.assertEquals("Ax", out.toString());
    }

    @SuppressWarnings("serial")
	@Test
    public void testSynchronizeSynchronizedProcess() {
    	final StringBuffer out = new StringBuffer();
    	executeProcess(new RunnableProcess() {
			public void run() {
		        Synch c1 = Jacob.newChannel(Synch.class, "");
		        object(receive(c1, new Synch() {
					public void ret() {
						out.append("x");
					}
				}));
		        Synch c2 = Jacob.newChannel(Synch.class, "");
		        object(receive(c2, new Synch() {
					public void ret() {
						out.append("y");
					}
				}));
		        Synchronized process = synchronizedProcess(c1, "S", out);
				instance(sync(c2, process));
			}
        });
    	// Both return hooks "x" and "y" are executed after synchronized process "S" 
    	Assert.assertEquals("Sxy", out.toString());
    }

    @SuppressWarnings("serial")
	@Test
    public void testInterceptProcess() {
    	final StringBuffer out = new StringBuffer();
    	executeProcess(new RunnableProcess() {
			public void run() {
		        Synch c1 = Jacob.newChannel(Synch.class, "");
		        object(receive(c1, new Synch() {
					public void ret() {
						out.append("x");
					}
				}));
				instance(intercept(sync(c1, atomicProcess("A", out)), atomicProcess("B", out)));
			}
        });
    	// Return interceptor "B" is executed after process "A", but before the hook "x" 
    	Assert.assertEquals("ABx", out.toString());
    }

    @SuppressWarnings("serial")
	@Test
    public void testInterceptSynchronizedProcess() {
    	final StringBuffer out = new StringBuffer();
    	executeProcess(new RunnableProcess() {
			public void run() {
		        Synch c1 = Jacob.newChannel(Synch.class, "");
		        object(receive(c1, new Synch() {
					public void ret() {
						out.append("x");
					}
				}));
		        Synch c2 = Jacob.newChannel(Synch.class, "");
		        object(receive(c2, new Synch() {
					public void ret() {
						out.append("y");
					}
				}));
				instance(intercept(sync(c1, atomicProcess("A", out)), sync(c2, atomicProcess("B", out))));
			}
        });
    	// Return interceptor "B" is executed after process "A", but before the hook "x" 
    	Assert.assertEquals("AByx", out.toString());
    }

    @SuppressWarnings("serial")
    @Test
    public void testSimpleSequence() {
    	final StringBuffer out = new StringBuffer();
    	executeProcess(new RunnableProcess() {
			public void run() {
				Jacob.instance(sequence(
			        atomicProcess("A", out), 
			        atomicProcess("B", out), 
			        atomicProcess("C", out)));
			}
        });
    	// TODO: explain 
    	Assert.assertEquals("ABC", out.toString());
    }

    @SuppressWarnings("serial")
    @Test
    public void testSynchronizedSequence() {
    	final StringBuffer out = new StringBuffer();
    	executeProcess(new RunnableProcess() {
			public void run() {
		        Synch c1 = Jacob.newChannel(Synch.class, "");
		        object(receive(c1, new Synch() {
					public void ret() {
						out.append("x");
					}
				}));
				Jacob.instance(sequence(c1,
			        atomicProcess("A", out), 
			        atomicProcess("B", out), 
			        atomicProcess("C", out), 
			        atomicProcess("D", out)));
			}
        });
    	// TODO: explain 
    	Assert.assertEquals("ABCDx", out.toString());
    }

    @SuppressWarnings("serial")
    @Test
    public void testTransitiveSequence() {
    	final StringBuffer out = new StringBuffer();
    	executeProcess(new RunnableProcess() {
			public void run() {
		        Synch c1 = Jacob.newChannel(Synch.class, "");
		        object(receive(c1, new Synch() {
					public void ret() {
						out.append("x");
					}
				}));
				Jacob.instance(sequence(c1,
			        atomicProcess("A", out), 
			        sequence(atomicProcess("B", out), atomicProcess("C", out)), 
			        atomicProcess("D", out)));
			}
        });
    	// TODO: explain 
    	Assert.assertEquals("ABCDx", out.toString());
    }

    @SuppressWarnings("serial")
    @Test
    public void testSequenceComposition() {
    	final StringBuffer out = new StringBuffer();
    	executeProcess(new RunnableProcess() {
			public void run() {
		        Synch c1 = Jacob.newChannel(Synch.class, "");
		        object(receive(c1, new Synch() {
					public void ret() {
						out.append("x");
					}
				}));
		        Synch c2 = Jacob.newChannel(Synch.class, "");
		        object(receive(c2, new Synch() {
					public void ret() {
						out.append("y");
					}
				}));
		        // just test a more complex scenario once
				Jacob.instance(sequence(c1,
				    sequence(
				        sequence(
				        	atomicProcess("A", out),
				        	atomicProcess("B", out),
				        	atomicProcess("C", out)),
				        atomicProcess("D", out)), 
				    atomicProcess("E", out), 
			        sequence(c2,
			        	atomicProcess("F", out), 
		        	    sequence(
		        	    	atomicProcess("G", out),
        	                atomicProcess("H", out)))));
			}
        });
    	// TODO: explain 
    	Assert.assertEquals("ABCDEFGHxy", out.toString());
    }

    @SuppressWarnings("serial")
	protected RunnableProcess atomicProcess(final String id, final StringBuffer out) {
    	return new RunnableProcess() {
			public void run() {
				out.append(id);
			}
    	};
    }

    @SuppressWarnings("serial")
	protected Synchronized synchronizedProcess(final Synch callback, final String id, final StringBuffer out) {
    	return new Synchronized(callback) {
			public void execute() {
				out.append(id);
			}
    	};
    }
    
    protected void executeProcess(final RunnableProcess process) {
        final JacobVPU vpu = new JacobVPU();
        vpu.setContext(new ExecutionQueueImpl());
        vpu.inject(process);
        while (vpu.execute()) {
            // keep doing it...
        }
    }

}