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

import com.malhartech.bufferserver.util.SerializedData;
import com.malhartech.bufferserver.util.WaitingChannelFutureListener;
import io.netty.channel.Channel;

/**
 *
 * @author chetan
 */
public class PhysicalNode
{
  public static final int BUFFER_SIZE = 64 * 1024;
  private int writtenBytes;
  private final long starttime;
  private final Channel channel;
  private long processedMessageCount;

  /**
   *
   * @param channel
   */
  public PhysicalNode(Channel channel)
  {
    this.channel = channel;
    starttime = System.currentTimeMillis();
    processedMessageCount = 0;
  }

  /**
   *
   * @return long
   */
  public long getstartTime()
  {
    return starttime;
  }

  /**
   *
   * @return long
   */
  public long getUptime()
  {
    return System.currentTimeMillis() - starttime;
  }

  final WaitingChannelFutureListener wcfl = new WaitingChannelFutureListener();

  /**
   *
   * @param d
   */
  public void send(SerializedData d) throws InterruptedException
  {
    if (BUFFER_SIZE - writtenBytes < d.size) {
      channel.flush().await(15);
      writtenBytes = 0;
    }
    channel.write(d);
    writtenBytes += d.size;
    processedMessageCount++;
  }

  /**
   *
   * @return long
   */
  public final long getProcessedMessageCount()
  {
    return processedMessageCount;
  }

  /**
   *
   * @param o
   * @return boolean
   */
  @Override
  public boolean equals(Object o)
  {
    return o == this || (o.getClass() == this.getClass() && o.hashCode() == this.hashCode());
  }

  /**
   *
   * @return int
   */
  public final int getId()
  {
    return getChannel().id();
  }

  /**
   *
   * @return int
   */
  @Override
  public final int hashCode()
  {
    return getChannel().id();
  }

  /**
   * @return the channel
   */
  public Channel getChannel()
  {
    return channel;
  }

}

@@ -257,16 +257,73 @@
 Channel;
+%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;
 %0A%0A/**%0A *
@@ -1188,16 +1188,87 @@
 size) %7B%0A
+      logger.info(%22since wrote %7B%7D bytes - waiting now%22, writtenBytes);%0A
       ch
@@ -2043,10 +2043,94 @@
 l;%0A  %7D%0A%0A
+  private static final Logger logger = LoggerFactory.getLogger(PhysicalNode.class);%0A
 %7D%0A

debug statements.