<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
 
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.toto</groupId>
    <artifactId>server</artifactId>
    <packaging>ear</packaging>
    <version>0.0.1-SNAPSHOT</version>
    
    <dependencies>
 
        <dependency>
            <groupId>com.toto</groupId>
            <artifactId>ejb-forward</artifactId>
            <version>0.0.1-SNAPSHOT</version>
            <type>ejb</type>
        </dependency>
 
        <dependency>
            <groupId>com.toto</groupId>
            <artifactId>webstart-client</artifactId>
            <version>0.0.1-SNAPSHOT</version>
            <type>war</type>
        </dependency>
 
    </dependencies>
 
    <build>
        <plugins>
 
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-ear-plugin</artifactId>
                <version>2.4.2</version>
                <configuration>
                    <version>5</version>
                    <modules>
 
                        <ejbModule>
                            <groupId>com.toto</groupId>
                            <artifactId>ejb-forward</artifactId>
                        </ejbModule>
 
                        <webModule>
                            <groupId>com.toto</groupId>
                            <artifactId>webstart-client</artifactId>
                            <contextRoot>/client</contextRoot>
                        </webModule>
 
                    </modules>
                </configuration>
            </plugin>
        </plugins>
        
    </build>
</project>

import java.io.IOException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
/**
 * Producer-Consumer in Java, for J2SE 1.5 using concurrent.
 */
public class ProdCons {
 
    private static ExecutorService pool = Executors.newCachedThreadPool();
    
    static boolean killProducers = false;
    static boolean killConsumers = false;
    static int queueSize = 1000;
    /**
     * Launch nP producers and nC consumers
     * 
     * @param nP
     * @param nC
     * @throws InterruptedException
     */
    ProdCons(int nP, int nC) throws InterruptedException {
 
        BlockingQueue<Long> fifo = new ArrayBlockingQueue<Long>(queueSize);
 
        for (int i = 0; i < nP; i++){
            Producer p = new Producer(fifo, 0.1f);
            pool.execute(p);
        }
        for (int i = 0; i < nC; i++){
            Consumer c = new Consumer(fifo);
            c.setDaemon(true);
            pool.execute(c);
        }
 
    }
    
    public static void main(String[] args) throws IOException,
            InterruptedException {
 
        // Start producers and consumers
        int numProducers = 1;
        int maxConsumers = 4; // max number of consumers
        ProdCons pc = new ProdCons(numProducers, maxConsumers);
 
        // Run for 5 seconds - shut down gracefully
        Thread.sleep(2000);
        pool.shutdown();
        killProducers = true;
        killConsumers = true;
    }
}

import java.util.Random;
import java.util.concurrent.BlockingQueue;
 
public class Producer implements Runnable {
 
    protected BlockingQueue<Long> queue;
 
    // Time interval between two polls of the database
    protected int pollInterval;
 
    private Random generator = new Random();
 
    Producer(BlockingQueue<Long> theQueue, float pollInterval) {
        this.queue = theQueue;
        setPollInterval(pollInterval);
    }
 
    public float getPollInterval() {
        return (pollInterval / 1000);
    }
 
    public void setPollInterval(float pollInterval) {
        this.pollInterval = java.lang.Math.round(pollInterval * 1000f);
    }
 
    public void run() {
        try {
            while (!ProdCons.killProducers) {
                Long[] justRead = requestDatabase();
                for (int i = 0; i < justRead.length; i++) {
                    Long id = justRead[i];
                    if (queue.remainingCapacity() > 0)
                        queue.put(id);
                }
                System.out.println("Produced " + justRead.length
                        + " object; List size now " + queue.size());
 
                // wait before next read in database
                Thread.sleep(pollInterval);
            }
        } catch (InterruptedException ex) {
        } finally {
            System.out.println("PRODUCER INTERRUPTED" );
            Thread.currentThread().interrupt();
        }
    }
 
    /**
     * Simulation of reading from database
     * 
     * @return
     */
    private Long[] requestDatabase() {
        // simulate retrieval of id in DB
        Long[] batch = new Long[generator.nextInt(1000)];
        for (int i = 0; i < batch.length; i++) {
            batch[i] = new Long(generator.nextInt());
        }
 
        try {
            Thread.sleep(100); // simulate time passed during query
        } catch (InterruptedException ex) {
            System.out.println("Producer Read INTERRUPTED" );
        }
        
        return (batch);
    }
 
}

import java.util.concurrent.BlockingQueue;
 
public class Consumer extends Thread implements Runnable {
    protected BlockingQueue<Long> queue;
 
    static long step;
 
    Consumer(BlockingQueue<Long> theQueue) {
        this.queue = theQueue;
    }
 
    public void run() {
        try {
            while (!ProdCons.killConsumers) {
                if(!queue.isEmpty()){
                    Long id = queue.take();
                    process(id);
                    if (++step % 1000 == 0)
                        System.out.println(step + " : FIFO size = " + queue.size());
                }
            }
        } catch (InterruptedException ex) {
        } finally {
            Thread.currentThread().interrupt();
            System.out.println("CONSUMER INTERRUPTED" );
        }
    }
 
    void process(Long id) {
        // Thread.sleep(xxx) // Simulate time passing
         if (step % 10000 == 0)
             System.out.println("Consuming id " + id);
    }
}

import java.io.IOException;
import java.util.concurrent.*;
 
/**
 * Producer-Consumer in Java, for J2SE 1.5 using concurrent.
 */
public class ProdCons {
    private static ExecutorService pool = Executors.newFixedThreadPool(5);
    static int queueSize = 1000;
 
    /**
     * Launch nP producers and nC consumers
     *
     * @param nP
     * @param nC
     * @throws InterruptedException
     */
    ProdCons(int nP, int nC) throws InterruptedException {
        BlockingQueue<Long> fifo = new ArrayBlockingQueue<Long>(queueSize);
        for (int i = 0; i < nP; i++) {
            Producer p = new Producer(fifo, 0.1f);
            pool.execute(p);
        }
        for (int i = 0; i < nC; i++) {
            Consumer c = new Consumer(fifo);
            pool.execute(c);
        }
    }
 
    public static void main(String[] args) throws IOException,
            InterruptedException {
        // Start producers and consumers
        int numProducers = 1;
        int maxConsumers = 4; // max number of consumers
        ProdCons pc = new ProdCons(numProducers, maxConsumers);
        // Run for 5 seconds - shut down gracefully
        Thread.sleep(5000);
        System.out.println("SHUT DOWN EVERYTHING" );
        pool.shutdownNow();
        System.out.println("SHUTTING DOWN EVERYTHING" );
        pool.awaitTermination(5, TimeUnit.DAYS);
        System.out.println("MADAGASCAR IS SAFE AGAIN" );
    }
}

import java.util.Random;
import java.util.concurrent.BlockingQueue;
 
public class Producer implements Runnable {
    protected BlockingQueue<Long> queue;
// Time interval between two polls of the database
    protected int pollInterval;
    private Random generator = new Random();
 
    Producer(BlockingQueue<Long> theQueue, float pollInterval) {
        this.queue = theQueue;
        setPollInterval(pollInterval);
    }
 
    public float getPollInterval() {
        return (pollInterval / 1000);
    }
 
    public void setPollInterval(float pollInterval) {
        this.pollInterval = java.lang.Math.round(pollInterval * 1000f);
    }
 
    public void run() {
        try {
            for (; ;) {
                final Long[] justRead = requestDatabase();
                for (final Long id : justRead) {
                    queue.put(id);
                }
                System.out.println("Produced " + justRead.length
                        + " object; List size now " + queue.size());
                // wait before next read in database
                Thread.sleep(pollInterval);
            }
        } catch (InterruptedException ex) {
            System.out.println("Producer interruption" );
        } finally {
            System.out.println("PRODUCER DYING" );
        }
    }
 
    /**
     * Simulation of reading from database
     *
     * @return
     */
    private Long[] requestDatabase() {
        // simulate retrieval of id in DB
        Long[] batch = new Long[generator.nextInt(1000)];
        for (int i = 0; i < batch.length; i++) {
            batch[i] = new Long(generator.nextInt());
        }
        try {
            Thread.sleep(100); // simulate time passed during query
        } catch (InterruptedException ex) {
            System.out.println("Producer Read INTERRUPTED" );
        }
        return (batch);
    }
}

import java.util.concurrent.BlockingQueue;
 
public class Consumer implements Runnable {
    protected BlockingQueue<Long> queue;
    static long step;
 
    Consumer(BlockingQueue<Long> theQueue) {
        this.queue = theQueue;
    }
 
    public void run() {
        try {
            for(;;) {
                Long id = queue.take();
                process(id);
                if (++step % 1000 == 0)
                    System.out.println(step + " : FIFO size = " + queue.size());
            }
        } catch (InterruptedException ex) {
            System.out.println("Consumer interruption" );
        } finally {
            System.out.println("CONSUMER DYING" );
        }
    }
 
    void process(Long id) {
        // Thread.sleep(xxx) // Simulate time passing
        if (step % 10000 == 0)
            System.out.println("Consuming id " + id);
    }
}

import java.io.IOException;
import java.util.List;
import java.util.concurrent.*;
 
/**
 * Producer-Consumer in Java, for J2SE 1.5 using concurrent.
 */
public class ProdCons {
    private static ExecutorService producer = Executors.newSingleThreadExecutor();
    private static ExecutorService consumers = Executors.newFixedThreadPool(10);
    
    public static void main(String[] args) throws IOException,
            InterruptedException {
        // Start producers and consumers
        Producer p = new Producer(consumers, 0.1f);
        producer.execute(p);
        
        // Run for 5 seconds
        Thread.sleep(5000);
        System.out.println("SHUT DOWN EVERYTHING" );
        producer.shutdownNow();
        final List<Runnable> stillRunnable = consumers.shutdownNow();
        System.out.println("SHUTTING DOWN EVERYTHING" );
        consumers.awaitTermination(5, TimeUnit.DAYS);
        System.out.println("MADAGASCAR IS SAFE AGAIN" );
        System.out.println(String.format("%d tasks not run", stillRunnable.size()));
    }
}

import java.util.Random;
import java.util.concurrent.Executor;
 
public class Producer implements Runnable {
// Time interval between two polls of the database
    protected int pollInterval;
    private Random generator = new Random();
    private Executor consumers;
 
    Producer(Executor consumers, float pollInterval) {
        this.consumers = consumers;
        setPollInterval(pollInterval);
    }
 
    public float getPollInterval() {
        return (pollInterval / 1000);
    }
 
    public void setPollInterval(float pollInterval) {
        this.pollInterval = java.lang.Math.round(pollInterval * 1000f);
    }
 
    public void run() {
        try {
            for (; ;) {
                final long[] justRead = requestDatabase();
                System.out.println("Producing " + justRead.length + " object" );
                for (final long id : justRead) {
                    consumers.execute(new Consumer(id));
                }
                // wait before next read in database
                Thread.sleep(pollInterval);
            }
        } catch (InterruptedException ex) {
            System.out.println("Producer interruption" );
        } finally {
            System.out.println("PRODUCER DYING" );
        }
    }
 
    /**
     * Simulation of reading from database
     *
     * @return
     */
    private long[] requestDatabase() {
        // simulate retrieval of id in DB
        long[] batch = new long[generator.nextInt(1000)];
        for (int i = 0; i < batch.length; i++) {
            batch[i] = generator.nextLong();
        }
        try {
            Thread.sleep(100); // simulate time passed during query
        } catch (InterruptedException ex) {
            System.out.println("Producer Read INTERRUPTED" );
        }
        return (batch);
    }
}

public class Consumer implements Runnable {
    static volatile long step = 0;
    private long id;
 
    Consumer(long id) {
        this.id = id;
    }
 
    public void run() {
        try {
            process(id);
            step++;
        } catch (InterruptedException ex) {
            System.out.println("Consumer interruption id " + id);
            // cancel operation and requeue here
        }
    }
 
    void process(Long id) throws InterruptedException {
        Thread.sleep(10);
        if (step % 1000 == 0)
            System.out.println("Consuming id " + id);
    }
}

import java.util.Collections;
import java.util.HashSet;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
 
public class Producer implements Runnable 
{
        
    // This set prevents that an id is inserted several times in the queue
    // This supposes that the rate of insertion is comparable to the rate of removal,
    // else the Set will grow up to memory exhaustion.
    protected static Set<Long> idToTreat = Collections.synchronizedSet(new HashSet<Long>(1000));
 
    private final ExecutorService consumerPool;
    
    // Time interval between two polls of the database
    private int pollInterval;
    private Random generator = new Random();
 
    /**
     * Create a Producer. 
     * Each producer has its own dedicated pool of consumers.
     * @param consumerPool
     * @param pollInterval
     */
    Producer(float pollInterval) {
        //The numbers are just silly tune parameters. Refer to the API.
        consumerPool = new ThreadPoolExecutor(1,5,30,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(100));
        setPollInterval(pollInterval);
    }
 
    public float getPollInterval() {
        return (pollInterval / 1000);
    }
 
    public void setPollInterval(float pollInterval) {
        this.pollInterval = java.lang.Math.round(pollInterval * 1000f);
    }
 
    public static Set<Long> getIdToTreat() {
        return idToTreat;
    }
 
    public static void setIdToTreat(Set<Long> idToTreat) {
        Producer.idToTreat = idToTreat;
    }
 
    public void run() {
        try {
            while (!Thread.interrupted()) {
                Long[] justRead = requestDatabase();
                for (int i = 0; i < justRead.length; i++) {
                    Long id = justRead[i];
                    consumerPool.submit(new Consumer(id));
                }
                
                // wait before next read in database
                Thread.sleep(pollInterval);
            }
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
            System.out.println("Producer INTERRUPTED" );
        }
    }
 
    /**
     * Simulation of reading from database
     * @return
     */
    private Long[] requestDatabase() { 
 
        // simulate reading of id in DB
        Long[] batch = new Long[generator.nextInt(100)];
        for (int i = 0; i < batch.length; i++) {
            batch[i] = new Long(generator.nextInt());
        }
 
        try {
            Thread.sleep(100); // simulate time passing during read
        } catch (InterruptedException ex) {
            System.out.println("Producer Read INTERRUPTED" );
        }
        return (batch);
    }
    
}

public class Consumer implements Runnable {
      static volatile long step = 0;
      private long id;
 
     Consumer(long id) { this.id = id; }
     
       public void run() {
             process(id);
             if (++step % 1000 == 0)
                   System.out.println(step + " : FIFO size = " + queue.size());
       }
     
       void process(Long id) {
           // Thread.sleep(xxx) // Simulate time passing
           if (step % 10000 == 0)
               System.out.println("Consuming id " + id);
       }
    }

public static String XMLEncNA(String s) {
    int ln = s.length();
    for (int i = 0; i < ln; i++) {
        char c = s.charAt(i);
        if (c == '<' || c == '>' || c == '&' || c == '"') {
            int next = 0;
            StringBuilder b = new StringBuilder();
            while (i < ln) {
                c = s.charAt(i);
                if (c == '<' || c == '>' || c == '&' || c == '"') {
                    b.append(s.substring(next, i));
                    switch (c) {
                        case '<': b.append("&lt;" ); break;
                        case '>': b.append("&gt;" ); break;
                        case '&': b.append("&amp;" ); break;
                        case '"': b.append("&quot;" ); break;
                    }
                    next = i + 1;
                }
                i++;
            }
            if (next < ln) b.append(s.substring(next));
            s = b.toString();
            break;
        } // if c ==
    } // for
    return s;
}