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

import com.google.protobuf.ByteString;
import com.malhartech.bufferserver.Buffer.Data;
import com.malhartech.bufferserver.Buffer.Data.DataType;
import com.malhartech.bufferserver.Buffer.PurgeRequest;
import com.malhartech.bufferserver.Buffer.SimpleData;
import com.malhartech.bufferserver.Buffer.SubscriberRequest;
import com.malhartech.bufferserver.policy.*;
import com.malhartech.bufferserver.util.SerializedData;
import io.netty.buffer.MessageBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.util.AttributeKey;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.slf4j.LoggerFactory;

/**
 * Handler to serve connections accepted by the server<p>
 * <br>
 *
 * @author <NAME> <<EMAIL>>
 */
@Sharable
public class ServerHandler extends ChannelInboundHandlerAdapter implements ChannelInboundMessageHandler<Data>
{
  private static final org.slf4j.Logger logger = LoggerFactory.getLogger(ServerHandler.class);
  private static final AttributeKey<DataList> DATALIST = new AttributeKey<DataList>("ServerHandler.datalist");
  private static final AttributeKey<LogicalNode> LOGICALNODE = new AttributeKey<LogicalNode>("ServerHandler.logicalnode");
  final HashMap<String, DataList> publisher_bufffers = new HashMap<String, DataList>();
  final HashMap<String, LogicalNode> groups = new HashMap<String, LogicalNode>();
  final ConcurrentHashMap<String, Channel> publisher_channels = new ConcurrentHashMap<String, Channel>();
  final ConcurrentHashMap<String, Channel> subscriber_channels = new ConcurrentHashMap<String, Channel>();

  @Override
  public final void inboundBufferUpdated(ChannelHandlerContext ctx) throws Exception
  {
    DataList dl = ctx.attr(DATALIST).get();

    MessageBuf<Data> in = ctx.inboundMessageBuffer();
    logger.debug("InboundBuffer updated with {} messages", in.size());
    for (int i = in.size(); i-- > 0;) {
      Data data = in.poll();
      switch (data.getType()) {
        case PUBLISHER_REQUEST:
          logger.info("Received publisher request: {}", data);
          dl = handlePublisherRequest(data.getPublishRequest(), ctx, data.getWindowId());
          break;

        case SUBSCRIBER_REQUEST:
          logger.info("Received subscriber request: {}", data);
          handleSubscriberRequest(data.getSubscribeRequest(), ctx, data.getWindowId());
          break;

        case PURGE_REQUEST:
          logger.info("Received purge request: {}", data);
          handlePurgeRequest(data.getPurgeRequest(), ctx, data.getWindowId());
          break;

        default:
          if (dl == null) {
            logger.error("Received packet {} when there is no datalist defined", data);
          }
          else {
            dl.add(data);
          }
          break;
      }
    }<commit_msg>less debugging.<commit_after>/*
 *  Copyright (c) 2012 Malhar, Inc.
 *  All Rights Reserved.
 */
package com.malhartech.bufferserver;

import com.google.protobuf.ByteString;
import com.malhartech.bufferserver.Buffer.Data;
import com.malhartech.bufferserver.Buffer.Data.DataType;
import com.malhartech.bufferserver.Buffer.PurgeRequest;
import com.malhartech.bufferserver.Buffer.SimpleData;
import com.malhartech.bufferserver.Buffer.SubscriberRequest;
import com.malhartech.bufferserver.policy.*;
import com.malhartech.bufferserver.util.SerializedData;
import io.netty.buffer.MessageBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.util.AttributeKey;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.slf4j.LoggerFactory;

/**
 * Handler to serve connections accepted by the server<p>
 * <br>
 *
 * @author <NAME> <<EMAIL>>
 */
@Sharable
public class ServerHandler extends ChannelInboundHandlerAdapter implements ChannelInboundMessageHandler<Data>
{
  private static final org.slf4j.Logger logger = LoggerFactory.getLogger(ServerHandler.class);
  private static final AttributeKey<DataList> DATALIST = new AttributeKey<DataList>("ServerHandler.datalist");
  private static final AttributeKey<LogicalNode> LOGICALNODE = new AttributeKey<LogicalNode>("ServerHandler.logicalnode");
  final HashMap<String, DataList> publisher_bufffers = new HashMap<String, DataList>();
  final HashMap<String, LogicalNode> groups = new HashMap<String, LogicalNode>();
  final ConcurrentHashMap<String, Channel> publisher_channels = new ConcurrentHashMap<String, Channel>();
  final ConcurrentHashMap<String, Channel> subscriber_channels = new ConcurrentHashMap<String, Channel>();

  @Override
  public final void inboundBufferUpdated(ChannelHandlerContext ctx) throws Exception
  {
    DataList dl = ctx.attr(DATALIST).get();

    MessageBuf<Data> in = ctx.inboundMessageBuffer();
//    logger.debug("InboundBuffer updated with {} messages", in.size());
    for (int i = in.size(); i-- > 0;) {
      Data data = in.poll();
      switch (data.getType()) {
        case PUBLISHER_REQUEST:
          logger.info("Received publisher request: {}", data);
          dl = handlePublisherRequest(data.getPublishRequest(), ctx, data.getWindowId());
          break;

        case SUBSCRIBER_REQUEST:
          logger.info("Received subscriber request: {}", data);
          handleSubscriberRequest(data.getSubscribeRequest(), ctx, data.getWindowId());
          break;

        case PURGE_REQUEST:
          logger.info("Received purge request: {}", data);
          handlePurgeRequest(data.getPurgeRequest(), ctx, data.getWindowId());
          break;

        default:
          if (dl == null) {
            logger.error("Received packet {} when there is no datalist defined", data);
          }
          else {
            dl.add(data);
          }
          break;
      }
    }
{'fraction_non_alphanumeric': 0.1131800518134715, 'fraction_numerical': 0.0022668393782383418, 'mean_word_length': 4.083950617283951, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2563268', 'n_tokens_mistral': 1794, 'n_tokens_neox': 1755, 'n_words': 393}