5
\$\begingroup\$

How can I improve this code?

Also available from git://github.com/edescourtis/actor.git .

Actor.java

package com.benbria.actor;


public interface Actor<T> extends Runnable {
    public abstract void send(T msg) throws InterruptedException;
    public abstract void stop() throws InterruptedException;
}

ActorListener.java

package com.benbria.actor;

public interface ActorListener<T> {
    void start(Actor<T> actor);
    void stop(Actor<T> actor);
    void exception(Actor<T> actor, Exception ex);   
}

Behaviour.java

package com.benbria.actor;

public interface Behaviour<T> {
    void receive(Actor<T> self, T msg);
}

ThreadActor.java

package com.benbria.actor;

import java.util.Comparator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;

import com.benbria.actor.listeners.NullActorListener;

/**
 * @author Eric des Courtis
 *
 */

public final class ThreadActor<T> implements Runnable, Actor<T> {
    private final BlockingQueue<StopOrT<T>> queue;
    private final Behaviour<T> behaviour;
    private final ActorListener<T> listener;

    public static <T> Actor<T> create(Behaviour<T> behaviour, ActorListener<T> listener) {
        return new ThreadActor<T>(behaviour, listener);
    }

    public static <T> Actor<T> create(Behaviour<T> behaviour) {
        return create(behaviour, new NullActorListener<T>());
    }

    public static <T> Actor<T> spawn(Behaviour<T> behaviour, ActorListener<T> listener){
        Actor<T> a = create(behaviour, listener);
        new Thread(a).start();
        return a;
    }

    public static <T> Actor<T> spawn(Behaviour<T> behaviour) {
        return spawn(behaviour, new NullActorListener<T>());
    }

    public static <T> Actor<T> createWithPriorityQueue(Behaviour<T> behaviour, ActorListener<T> listener, final Comparator<T> comparator) {
        return new ThreadActor<T>(behaviour, listener, new PriorityBlockingQueue<StopOrT<T>>(10, new Comparator<StopOrT<T>>() {
            @Override
            public int compare(StopOrT<T> o1, StopOrT<T> o2) {
                if(o1.isStop()) return -1;
                if(o2.isStop()) return 1;
                return comparator.compare(o1.getT(), o2.getT());
            }
        }));
    }

    public static <T> Actor<T> createWithPriorityQueue(Behaviour<T> behaviour, Comparator<T> comparator) {
        return createWithPriorityQueue(behaviour, new NullActorListener<T>(), comparator);
    }

    public static <T> Actor<T> spawnWithPriorityQueue(Behaviour<T> behaviour, ActorListener<T> listener, Comparator<T> comparator){
        Actor<T> a = createWithPriorityQueue(behaviour, listener, comparator);
        new Thread(a).start();
        return a;
    }

    public static <T> Actor<T> spawnWithPriorityQueue(Behaviour<T> behaviour, Comparator<T> comparator) {
        return spawnWithPriorityQueue(behaviour, new NullActorListener<T>(), comparator);
    }


    private ThreadActor(Behaviour<T> behaviour, ActorListener<T> listener, BlockingQueue<StopOrT<T>> queue) {
        this.listener = listener;
        this.behaviour = behaviour;
        this.queue = queue;
    }

    private ThreadActor(Behaviour<T> behaviour, ActorListener<T> listener) {
        this(behaviour, listener, new LinkedBlockingQueue<StopOrT<T>>());
    }

    private ThreadActor(Behaviour<T> behaviour) {
        this(behaviour, new NullActorListener<T>());
    }

    public void run() {
        listener.start(this);
        try {
            while(true) {
                StopOrT<T> stopOrMsg = queue.take();
                if(stopOrMsg.isT()){
                    behaviour.receive(this, stopOrMsg.getT());
                }else{
                    break;
                }
            }
            listener.stop(this);
        } catch (Exception ex) {
            listener.exception(this, ex);
        }
    }

    @Override
    public void send(T msg) throws InterruptedException {
        queue.put(StopOrT.newT(msg));
    }

    @Override
    public void stop() throws InterruptedException {
        StopOrT<T> stop = StopOrT.newStop();
        queue.put(stop);
    }   

    private static class StopOrT<T> {
        private final T t;
        private StopOrT(T t) {
            this.t = t;
        }

        public boolean isStop() {
            return t == null;
        }

        public boolean isT() {
            return t != null;
        }
        public T getT() {
            if (t == null)
                throw new RuntimeException("not T");
            return t;
        }
        public static <T> StopOrT<T> newStop() {
            return new StopOrT<T>(null);
        }
        public static <T> StopOrT<T> newT(T t) {
            return new StopOrT<T>(t);
        }
    }
}

NullActorListener.java

package com.benbria.actor.listeners;

import com.benbria.actor.Actor;
import com.benbria.actor.ActorListener;

public class NullActorListener<T> implements ActorListener<T> {
    public NullActorListener() { }

    @Override
    public void start(Actor<T> actor) { }

    @Override
    public void stop(Actor<T> actor) { }

    @Override
    public void exception(Actor<T> actor, Exception ex) { }

}

NullBehaviour.java

package com.benbria.actor.behaviours;

import com.benbria.actor.Actor;
import com.benbria.actor.Behaviour;

public final class NullBehaviour<T> implements
Behaviour<T> {
    @Override
    public void receive(Actor<T> self, T msg) {
    }
}

ThreadActorTests.java

/**
 * 
 */
package com.benbria.actor.tests;

import static org.junit.Assert.*;

import org.junit.Test;

import com.benbria.actor.Actor;
import com.benbria.actor.ActorListener;
import com.benbria.actor.Behaviour;
import com.benbria.actor.ThreadActor;
import com.benbria.actor.behaviours.NullBehaviour;


/**
 * @author Eric des Courtis
 *
 */
public class ThreadActorTests {
    class StartStopExceptionCheckingActorListener implements ActorListener<String> {
        private boolean started = false;
        private boolean stopped = false;
        private boolean exception = false;
        @Override
        public void start(Actor<String> actor) {
            setStarted();
        }

        @Override
        public void stop(Actor<String> actor) {
            setStopped();
        }

        @Override
        public void exception(Actor<String> actor, Exception ex) {
            setException();
        }

        public boolean isStarted() {
            return started;
        }

        public void setStarted() {
            this.started = true;
        }

        public boolean isStopped() {
            return stopped;
        }

        public void setStopped() {
            this.stopped = true;
        }

        public boolean isException() {
            return exception;
        }

        public void setException() {
            this.exception = true;
        }
    };

    @Test
    public void testStartStopException(){
        StartStopExceptionCheckingActorListener checkListener = new StartStopExceptionCheckingActorListener();
        Actor<String> a = ThreadActor.spawn(new NullBehaviour<String>(), checkListener);

        try {
            a.stop();
            Thread.sleep(1000);
        } catch (InterruptedException e) {
        }


        assertTrue(checkListener.isStarted());
        assertTrue(checkListener.isStopped());
        assertFalse(checkListener.isException());

    }

    class ExceptionGeneratingBehaviour implements Behaviour<String> {
        @Override
        public void receive(Actor<String> self, String msg) {
            throw new RuntimeException("evil");
        }   
    }

    @Test
    public void testException() {
        StartStopExceptionCheckingActorListener checkListener = new StartStopExceptionCheckingActorListener();
        Actor<String> a = ThreadActor.spawn(new ExceptionGeneratingBehaviour(), checkListener);
        try {
            a.send("hello");
            Thread.sleep(1000);
            a.stop();
        } catch (InterruptedException e) {

        }

        assertTrue(checkListener.isException());
    }

    class ReceivedMessageCheckingBehaviour implements Behaviour<String> {
        private String msg;

        @Override
        public void receive(Actor<String> self, String msg) {
            this.msg = msg;
        }

        public String getMsg() {
            return msg;
        }
    }

    @Test
    public void testSend() {
        ReceivedMessageCheckingBehaviour behaviour = new ReceivedMessageCheckingBehaviour();
        Actor<String> a = ThreadActor.spawn(behaviour);
        try {
            a.send("testing");
            Thread.sleep(1000);
            a.stop();
        } catch (InterruptedException e) {
        }

        assertEquals("testing", behaviour.getMsg());

    }

}

LoggingBehaviour.java

package com.benbria.actor.behaviours;

import java.util.logging.Logger;

import com.benbria.actor.Actor;
import com.benbria.actor.Behaviour;

public final class LoggingBehaviour<T> implements
Behaviour<T> {
    private final static Logger logger = Logger.getLogger(LoggingBehaviour.class.getName());
    @Override
    public void receive(Actor<T> self, T msg) {
        logger.info(self + ": " + msg);
    }
}

LoggingActorListener.java

package com.benbria.actor.listeners;

import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.util.logging.Logger;

import com.benbria.actor.Actor;
import com.benbria.actor.ActorListener;

public class LoggingActorListener<T> implements ActorListener<T> {
    private static final Logger logger = Logger.getLogger(LoggingActorListener.class.getName());

    public LoggingActorListener() { }

    @Override
    public void start(Actor<T> actor) {
        logger.info(actor + ": started");
    }

    @Override
    public void stop(Actor<T> actor) {
        logger.info(actor + ": stopped");

    }

    @Override
    public void exception(Actor<T> actor, Exception ex) {

        ByteArrayOutputStream baos = new ByteArrayOutputStream(); 
        ex.printStackTrace(new PrintStream(baos, true));

        logger.severe(actor + ": " + baos.toString());
    }

}

BroadcastBehaviour.java

package com.benbria.actor.behaviours;

import java.util.LinkedList;
import java.util.List;

import com.benbria.actor.Actor;
import com.benbria.actor.Behaviour;

public class BroadcastBehaviour<T> implements Behaviour<T> {
    private List<Behaviour<T>> behaviourList = new LinkedList<Behaviour<T>>();

    public BroadcastBehaviour(List<Behaviour<T>> behaviourList) {
        this.behaviourList.addAll(behaviourList);
    }

    @Override
    public void receive(Actor<T> self, T msg) {
        for(Behaviour<T>  behaviour: behaviourList){
            behaviour.receive(self, msg);
        }
    }
}

BroadcastActorListener.java

package com.benbria.actor.listeners;

import java.util.LinkedList;
import java.util.List;

import com.benbria.actor.Actor;
import com.benbria.actor.ActorListener;

public class BroadcastActorListener<T> implements ActorListener<T> {
    private final List<ActorListener<T>> listeners = new LinkedList<ActorListener<T>>();

    public BroadcastActorListener(List<ActorListener<T>> listeners) {
        this.listeners.addAll(listeners);
    }

    @Override
    public void start(Actor<T> actor) {
        for(ActorListener<T> listener: listeners) {
            listener.start(actor);
        }
    }

    @Override
    public void stop(Actor<T> actor) {
        for(ActorListener<T> listener: listeners) {
            listener.stop(actor);
        }       
    }

    @Override
    public void exception(Actor<T> actor, Exception ex) {
        for(ActorListener<T> listener: listeners) {
            listener.exception(actor, ex);
        }
    }

}
\$\endgroup\$
  • \$\begingroup\$ Very nice! What about an factory to create the actor? \$\endgroup\$ – megazord Apr 17 '13 at 21:56
  • \$\begingroup\$ Not sure if it's worth doing. Anyone else have input on this? \$\endgroup\$ – Eric des Courtis Apr 17 '13 at 21:57
  • \$\begingroup\$ ThreadActor is a factory (having static create() methods). \$\endgroup\$ – Alexei Kaigorodov Apr 19 '13 at 9:04
2
\$\begingroup\$
  1. No need for Actor to extend Runnable

  2. NullActorListener is redundant: MultiplexedActorListener with empty list of listeners behaves the same way. Having MultiplexedActorListener as a default listener, no need for the ThreadActor constructors with ActorListener parameters - one can just add listeners after ThreadActor creation.

  3. Multiplexer is a wrong term, you meant demultiplexer, see Wikipedia

  4. Parameter Actor self in Behaviour.receive() can be dropped. The case when Behaviour really needs access to its own actor is rare, and can be implemented with passing Actor to Behaviour implementation constructor.

| improve this answer | |
\$\endgroup\$
  • 1
    \$\begingroup\$ 3. You are right that Multiplexer is the wrong term but come to think of it Demultiplexer isn't either. I changed it to Broadcast instead. 4. The parameter cannot be dropped because the actor needs to be able to tell others about itself. \$\endgroup\$ – Eric des Courtis Apr 19 '13 at 18:48
  • 1
    \$\begingroup\$ 1. Actor needs to extend Runnable because in some containers where threads are managed you must create the thread on your own and register it with the container. If the actor does not implements Runnable this is not possible. 2. I simply don't agree that they are equivalent. The meaning is lost if the name is changed regardless if they produce the same output. \$\endgroup\$ – Eric des Courtis Apr 19 '13 at 18:52

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.