Start line:  
End line:  

Snippet Preview

Snippet HTML Code

Stack Overflow Questions
BEGIN LICENSE BLOCK ***** Version: EPL 1.0/GPL 2.0/LGPL 2.1 The contents of this file are subject to the Eclipse Public License Version 1.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.eclipse.org/legal/epl-v10.html Software distributed under the License is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License for the specific language governing rights and limitations under the License. Copyright (C) 2008 MenTaLguY <mental@rydia.net> Alternatively, the contents of this file may be used under the terms of either of the GNU General Public License Version 2 or later (the "GPL"), or the GNU Lesser General Public License Version 2.1 or later (the "LGPL"), in which case the provisions of the GPL or the LGPL are applicable instead of those above. If you wish to allow use of your version of this file only under the terms of either the GPL or the LGPL, and not to allow others to use your version of this file under the terms of the EPL, indicate your decision by deleting the provisions above and replace them with the notice and other provisions required by the GPL or the LGPL. If you do not delete the provisions above, a recipient may use your version of this file under the terms of any one of the EPL, the GPL or the LGPL. END LICENSE BLOCK ***
 
 package org.jruby.util;
 
 
 public final class Join {
     public static final Executor TRIVIAL_EXECUTOR = new Executor() {
         public void execute(Runnable command) {
             (new Thread(command)).start();
         }
     };
 
     private final Executor executor;
     private final LinkedList[] writes;
     private final long asyncMask;
     private long mask = 0;
     private final Reaction[][] reactionsPerChannel;
 
     public static class Spec {
         private ArrayList<ArrayList<Reaction>> reactionsPerChannel = new ArrayList<ArrayList<Reaction>>();
         private long asyncMask = 0;
         private long mask = 0;
         private volatile Reaction[][] cachedReactionsPerChannel = null;
 
         public Spec() {}
 
         public void addReaction(Reaction reaction) {
             if ( (  & ~ & reaction.asyncMask ) != 0 ) {
                 throw new IllegalArgumentException("Cannot use a synchronous channel in a non-head position");
             }
             if ( ( reaction.mask & ~reaction.asyncMask &  ) != 0 ) {
                 throw new IllegalArgumentException("Cannot use an asynchronous channel in the head position of a synchronous reaction");
             }
              = null;
             final int[] indices = reaction.indices;
             for ( int i = 0 ; i < indices.length ; i++ ) {
                 final int index = indices[i];
                 if ( .size() <= index ) {
                     .ensureCapacity(index+1);
                     while ( .size() <= index ) {
                         .add(null);
                     }
                 }
                 ArrayList<Reactionreactions = .get(index);
                 if ( reactions == null ) {
                     reactions = new ArrayList<Reaction>();
                     .set(indexreactions);
                 }
                 reactions.add(reaction);
             }
              |= reaction.asyncMask;
              |= reaction.mask;
         }
 
         public Join createJoin() {
             return createJoin();
         }
 
         private static final Reaction[] EMPTY_REACTIONS = new Reaction[0];
         public Join createJoin(final Executor executor) {
             if ( == null) {
                 final int length = .size();
                 final Reaction[][] localReactionsPerChannel = new Reaction[length][];
                 for ( int i = 0 ; i < length ; ++i ) {
                     final ArrayList<Reactionreactions = .get(i);
                     if ( reactions != null ) {
                         localReactionsPerChannel[i] = reactions.toArray();
                     }
                 }
                  = localReactionsPerChannel;
             }
            return new Join(executor);
        }
    }
    public static abstract class Reaction {
        private final int[] indices;
        private final long mask;
        private final long asyncMask;
        private static int[] toIndices(Enum<?> headEnum<?>[] channels) {
            final int[] indices = new int[channels.length+1];
            indices[0] = head.ordinal();
            for ( int i = 0 ; i < channels.length ; ++i ) {
                indices[i+1] = channels[i].ordinal();
            }
            return indices;
        }
        Reaction(Enum<?> headEnum<?>[] channelsboolean isAsync) {
            this(toIndices(headchannels), isAsync);
        }
        Reaction(int[] indicesboolean isAsync) {
            long mask = 0;
            for ( int i = 0 ; i < indices.length ; ++i ) {
                final int index = indices[i];
                if ( index < 0 || index > 63 ) {
                    throw new IndexOutOfBoundsException();
                }
                if ( ( mask & ( 1L << index ) ) != 0 ) {
                    throw new IllegalArgumentException("Duplicate channels in reaction");
                }
                mask |= 1L << index;
            }
            this. = indices;
            this. = mask;
            if (isAsync) {
                this. = mask;
            } else {
                this. = mask & ~( 1L << indices[0] );
            }
        }
        abstract void dispatch(Join joinObject[] args);
    }
    public static abstract class FastReaction extends Reaction {
        public FastReaction(int[] indices) {
            super(indices.clone(), true);
        }
        public FastReaction(Enum<?> headEnum<?> ... channels) {
            super(headchannelstrue);
        }
        @Override
        void dispatch(final Join joinfinal Object[] args) {
            try {
                react(joinargs);
            } catch (Exception e) {
            }
        }
        public abstract void react(Join joinObject[] args);
    }
    public static abstract class AsyncReaction extends Reaction {
        public AsyncReaction(int[] indices) {
            super(indices.clone(), true);
        }
        public AsyncReaction(Enum<?> headEnum<?> ... channels) {
            super(headchannelstrue);
        }
        @Override
        void dispatch(final Join joinfinal Object[] args) {
            final AsyncReaction reaction = this;
            join.executor.execute(new Runnable() {
                public void run() {
                    reaction.react(joinargs);
                }
            });
        }
        public abstract void react(Join joinObject[] args);
    }
    public static abstract class SyncReaction extends Reaction {
        public SyncReaction(int[] indices) {
            super(indices.clone(), false);
        }
        public SyncReaction(Enum<?> headEnum<?> ... channels) {
            super(headchannelsfalse);
        }
        @Override
        void dispatch(Join joinfinal Object[] args) {
            final Call call = (Call)args[0];
            args[0] = call.getMessage();
            call.activate(jointhisargs); 
        }
        public abstract Object react(Join joinObject[] args);
    }
    private Join(final long asyncMaskfinal Reaction[][] reactionsPerChannelExecutor executor) {
        final LinkedList[] writes = new LinkedList[reactionsPerChannel.length];
        for ( int i = 0 ; i < writes.length ; ++i ) {
            if ( reactionsPerChannel[i] != null ) {
                writes[i] = new LinkedList();
            }
        }
        this. = asyncMask;
        this. = reactionsPerChannel;
        this. = writes;
        this. = executor;
    }
    private void sendRaw(int indexObject message) {
        Reaction selectedReaction = null;
        Object[] args = null;
        synchronized (this) {
            final LinkedList writing = [index];
            if ( writing == null ) {
                throw new IndexOutOfBoundsException();
            }
            writing.addLast(message);
             |= 1L << index;
            final Reaction[] reactions = [index];
            for (Reaction reactionreactions) {
                if ( ( reaction.mask &  ) == reaction.mask ) {
                    final int[] indices = reaction.indices;
                    args = new Object[indices.length];
                    for ( int i = 0 ; i < indices.length ; ++i ) {
                        final int readIndex = indices[i];
                        final LinkedList reading = [readIndex];
                        args[i] = reading.removeFirst();
                        if (reading.isEmpty()) {
                             &= ~(1L << readIndex);
                        }
                    }
                    selectedReaction = reaction;
                    break;
                }
            }
        }
        if ( selectedReaction != null ) {
            selectedReaction.dispatch(thisargs);
        }
    }
    public boolean isAsync(int channel) {
        return ( ( 1L << channel ) &  ) != 0;
    }
    public void send(int channelObject message) {
        if (isAsync(channel)) {
            sendRaw(channelmessage);
        } else {
            sendRaw(channelnew AsyncCall(message));
        }
    }
    public void send(Enum<?> channelObject message) {
        send(channel.ordinal(), message);
    }
    public Object call(int channelObject message) {
        if (isAsync(channel)) {
            sendRaw(channelmessage);
            return null;
        } else {
            SyncCall request = new SyncCall(message);
            sendRaw(channelrequest);
            return request.call();
        }
    }
    public Object call(Enum<?> channelObject message) {
        return call(channel.ordinal(), message);
    }
    private static abstract class Call {
        private final Object message;
        public Call(Object message) {
            this. = message;
        }
        public Object getMessage() {
            return ;
        }
        public abstract void activate(Join joinSyncReaction reactionObject[] args);
    }
    private static class AsyncCall extends Call {
        public AsyncCall(Object message) {
            super(message);
        }
        
        public void activate(final Join joinfinal SyncReaction reactionfinal Object[] args) {
            join.executor.execute(new Runnable() {
                public void run() {
                    reaction.react(joinargs);
                }
            }); 
        }
    }
    private static class SyncCall extends Call {
        private Join join = null;
        private SyncReaction reaction = null;
        private Object[] args = null;
        public SyncCall(Object message) {
            super(message);
        }
        
        public synchronized void activate(Join joinSyncReaction reactionObject[] args) {
            this. = join;
            this. = reaction;
            this. = args;
            notifyAll();
        }
        public synchronized Object call() {
            boolean interrupted = false;
            try {
                while (  == null ) {
                    try {
                        wait();
                    } catch (InterruptedException e) {
                        interrupted = true;
                    }
                }
            } finally {
                if (interrupted) {
                    Thread.currentThread().interrupt();
                }
            }
            return .react();
        }
    }
New to GrepCode? Check out our FAQ X