/*

  * 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.giraph.comm.messages;

 import it.unimi.dsi.fastutil.longs.LongArrayList;

 import it.unimi.dsi.fastutil.longs.LongListIterator;

 import java.io.IOException;

 import java.util.Iterator;

 import org.apache.giraph.factories.MessageValueFactory;

 import org.apache.giraph.utils.ExtendedByteArrayOutputBuffer;

 import org.apache.giraph.utils.ExtendedDataOutput;

 import org.apache.giraph.utils.UnsafeReusableByteArrayInput;

 import org.apache.hadoop.io.Writable;

 /**

  * Create an iterable for messages based on a pointer list

  *

  * @param <M> messageType

  */

 public class PointerListMessagesIterable<M extends Writable>

   implements Iterable<M> {

   /** Message class */

   private final MessageValueFactory<M> messageValueFactory;

   /** List of pointers to messages in byte array */

   private final LongArrayList pointers;

   /** Holds the byte arrays of serialized messages */

   private final ExtendedByteArrayOutputBuffer msgBuffer;

   /** Reader to read data from byte buffer */

   private final UnsafeReusableByteArrayInput messageReader;

   /**

    *

    * @param messageValueFactory message value factory

    * @param pointers pointers to messages in buffer

    * @param msgBuffer holds the byte arrays of serialized messages

    */

   public PointerListMessagesIterable(MessageValueFactory<M> messageValueFactory,

     LongArrayList pointers, ExtendedByteArrayOutputBuffer msgBuffer) {

     this.messageValueFactory = messageValueFactory;

     this.pointers = pointers;

     this.msgBuffer = msgBuffer;

     // TODO - if needed implement same for Safe as well

     messageReader = new UnsafeReusableByteArrayInput();

   }

   /**

    * Create message from factory

    *

    * @return message instance

    */

   protected M createMessage() {

     return messageValueFactory.newInstance();

   }

   @Override

   public Iterator<M> iterator() {

     return new Iterator<M>() {

       private final LongListIterator iterator = pointers.iterator();

       private final M reusableMsg =

         PointerListMessagesIterable.this.createMessage();

       @Override

       public boolean hasNext() {

         return iterator.hasNext();

       }

       @Override

       public M next() {

         long pointer = iterator.nextLong();

         try {

           int index = (int) (pointer >>> 32);

           int offset = (int) pointer;

           ExtendedDataOutput buffer = msgBuffer.getDataOutput(index);

           messageReader.initialize(buffer.getByteArray(), offset,

             buffer.getPos());

           reusableMsg.readFields(messageReader);

         } catch (IOException e) {

           throw new IllegalStateException("Got exception : " + e);

         }

         return reusableMsg;

       }

       @Override

       public void remove() {

         throw new UnsupportedOperationException();

       }

     };

   }

 }