/*
 *  Copyright (c) 2012 Malhar, Inc.
 *  All Rights Reserved.
 */
package com.malhartech.bufferserver.internal;

import com.malhartech.bufferserver.packet.MessageType;
import com.malhartech.bufferserver.util.SerializedData;
import java.util.Iterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author chetan
 */
class DataListIterator implements Iterator<SerializedData>
{
  private static final Logger logger = LoggerFactory.getLogger(DataListIterator.class);
  Block da;
  SerializedData previous = null;
  SerializedData current = new SerializedData();

  /**
   *
   * @param da
   */
  DataListIterator(Block da)
  {
    da.acquire(true);
    this.da = da;

    current.bytes = da.data;
    current.offset = da.readingOffset;
  }

  /**
   *
   * @return boolean
   */
  @Override
  public synchronized boolean hasNext()
  {
    while (true) {
      da.getNextData(current);
      switch (current.size) {
        case -1:
          return false;

        case 0:
          if (da.next == null) {
            return false;
          }

          da.release(false);
          da.next.acquire(true);
          da = da.next;
          current.bytes = da.data;
          current.offset = da.readingOffset;
          break;

        default:
          if (current.bytes[current.dataOffset] != MessageType.NO_MESSAGE_VALUE) {
              return true;
          }
          current.offset += current.size;
          break;
      }
    }
  }

  @Override
  @SuppressWarnings("FinalizeDeclaration")
  protected void finalize() throws Throwable
  {
    da.release(false);
    super.finalize();
  }

  /**
   *
   * @return {@link com.malhartech.bufferserver.util.SerializedData}
   */
  @Override
  public SerializedData next()
  {
    previous = current;
    current = new SerializedData();
    current.offset = previous.offset + previous.size;
    current.bytes = previous.bytes;
    return previous;
  }

  /**
   * Removes from the underlying collection the last element returned by the iterator (optional operation). This method can be called only once per call to
   * next. The behavior of an iterator is unspecified if the underlying collection is modified while the iteration is in progress in any way other than by
   * calling this method.
   */
  @Override
  public void remove()
  {
    if (previous == null) {
      throw new IllegalStateException("Nothing to remove");
    }

    previous.bytes[previous.dataOffset] = MessageType.NO_MESSAGE_VALUE;
  }

  void rewind(int processingOffset)
  {
    current.offset = processingOffset;
  }

}

@@ -163,16 +163,68 @@
 geType;%0A
+import com.malhartech.bufferserver.storage.Storage;%0A
 import c
@@ -628,16 +628,49 @@
 dData();
+%0A  private final Storage storage;
 %0A%0A  /**%0A
@@ -722,16 +722,33 @@
 Block da
+, Storage storage
 )%0A  %7B%0A  
@@ -752,32 +752,41 @@
 %0A    da.acquire(
+storage, 
 true);%0A    this.
@@ -793,16 +793,44 @@
 da = da;
+%0A    this.storage = storage;
 %0A%0A    cu
@@ -1213,32 +1213,41 @@
      da.release(
+storage, 
 false);%0A        
@@ -1264,16 +1264,25 @@
 acquire(
+storage, 
 true);%0A 
@@ -1738,16 +1738,25 @@
 release(
+storage, 
 false);%0A

integrate with storage.