Building the generic worker
Prev | Chapter 3. Building a Grid Computing platform with Rio | Next

## Building the generic worker

As explained before, our worker will `take` `Task` entries from the space, run the `execute()` method on the taken entry, and `write` the result in the space. The generic worker (`TaskServer`) will be in need of a proxy to the JavaSpaces service. You could code the logic required to do a lookup of the Jini service, but you are going to use the IoC (Inversion of Control) support provided by Rio. What it means is that Rio will inject the JavaSpace proxy into your JSB. All you need to do is to define a setter method that will be called by Rio and put the appropriate configuration in the OpString. This JSB will also need a transaction manager. IoC will also be used in order to have a proxy for this service. You will learn later on how to write the OpString. Let's focus on the code of the `TaskServer` class.

    public class TaskServer extends ServiceBeanAdapter {
        private Thread job;
        private JavaSpace space;
        private TransactionManager txnManager;
        private CounterWatch rate;
        private ThroughputWatch throughput;
        private StopWatch meanTime;
        private long TIMEOUT;
        private Logger logger = Logger.getLogger(getClass().getName());

        ...

        /** Called by Rio. */
        public void setSpace(JavaSpace space) {
            logger.info("Using space: " + space);
            this.space = space;
        }

        /** Called by Rio. */
        public void setTransactionManager(TransactionManager txnManager) {
            logger.info("Using transaction manager: " + txnManager);
            this.txnManager = txnManager;
        }
    }

Now we will override the `initialize()` in order to initialize our watches, get the value of the timeout (time for leases) to use when writing in the space and finally to create a `Runnable` job.

    public void initialize(ServiceBeanContext context) throws Exception {
        super.initialize(context);
        TIMEOUT = (Long) super.getConfiguration().getEntry("org.jini.rio.tutorial.grid", "timeout", Long.class);
        job = new Thread(new TaskJob());
        rate = new CounterWatch("rate");
        throughput = new ThroughputWatch("throughput");
        meanTime = new StopWatch("meanTime");
        super.getWatchRegistry().register(rate);
        super.getWatchRegistry().register(throughput);
        super.getWatchRegistry().register(meanTime);
    }

The start() method is overrided too in order to start our job:

    public Object start(final ServiceBeanContext context) throws Exception {
        Object o = super.start(context);
        job.start();
        return o;
    }

Most of the code in running in the `TaskJob` class:

    private class TaskJob implements Runnable {
        public void run() {
            try {
                logger.info("Shapshotting task...");
                // wait for services to come up
                while (space == null || txnManager == null)
                    Thread.sleep(500);
                Entry template = space.snapshot(new AbstractTask());
                while (true) {
                    Transaction txn = TransactionFactory.create(txnManager, Lease.FOREVER).transaction;
                    Task task = (Task) space.takeIfExists(template, txn, TIMEOUT);
                    if (task == null) {
                        txn.abort();
                        Thread.sleep(200);
                        continue;
                    }
                    logger.info("Processing task: " + task);
                    meanTime.startTiming();
                    Entry result = task.execute();
                    meanTime.stopTiming();
                    space.write(result, txn, TIMEOUT);
                    txn.commit();
                    rate.increment();
                    throughput.increment();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

The `run()` method of the `TaskJob` class makes sure there is a valid proxy to JavaSpaces (not null), then loops indefinitely. In this loop, the thread creates a transaction, takes a `Task` `Entry` if there is one available. If not (`task == null`) the transaction is aborted and it loops, otherwise it keeps going. The various watches are properly updated and the `execute()` method is run on the taken `Task`. The result is then written in the space and the transaction committed.

Prev | Up | Next
Chapter 3. Building a Grid Computing platform with Rio | Home | Building and running tasks on your grid