Review Board 1.7.22


Patch for Flume 2130

Review Request #15874 - Created Nov. 27, 2013 and updated

Ashish Paliwal
FLUME-2130
Reviewers
Flume
flume-git
Rebased patch for handling larger payload for SyslogUdpSource
Added test condition for handling of packet larger than 768 bytes
flume-ng-core/src/main/java/org/apache/flume/source/SyslogUDPSource.java
Revision 96a9e85 New Change
[20] 30 lines
[+20]
31
import org.apache.flume.conf.Configurable;
31
import org.apache.flume.conf.Configurable;
32
import org.apache.flume.conf.Configurables;
32
import org.apache.flume.conf.Configurables;
33
import org.apache.flume.source.SyslogUtils;
33
import org.apache.flume.source.SyslogUtils;
34
import org.jboss.netty.bootstrap.ConnectionlessBootstrap;
34
import org.jboss.netty.bootstrap.ConnectionlessBootstrap;
35
import org.jboss.netty.buffer.ChannelBuffer;
35
import org.jboss.netty.buffer.ChannelBuffer;
36
import org.jboss.netty.channel.Channel;
36
import org.jboss.netty.channel.*;
37
import org.jboss.netty.channel.ChannelHandlerContext;

   
38
import org.jboss.netty.channel.ChannelPipeline;

   
39
import org.jboss.netty.channel.ChannelPipelineFactory;

   
40
import org.jboss.netty.channel.Channels;

   
41
import org.jboss.netty.channel.MessageEvent;

   
42
import org.jboss.netty.channel.SimpleChannelHandler;

   
43
import org.jboss.netty.channel.socket.oio.OioDatagramChannelFactory;
37
import org.jboss.netty.channel.socket.oio.OioDatagramChannelFactory;
44

    
   
38

   
45
import org.slf4j.Logger;
39
import org.slf4j.Logger;
46
import org.slf4j.LoggerFactory;
40
import org.slf4j.LoggerFactory;
47

    
   
41

   
[+20] [20] 9 lines
[+20] [+] public class SyslogUDPSource extends AbstractSource
57
  private static final Logger logger = LoggerFactory
51
  private static final Logger logger = LoggerFactory
58
      .getLogger(SyslogUDPSource.class);
52
      .getLogger(SyslogUDPSource.class);
59

    
   
53

   
60
  private CounterGroup counterGroup = new CounterGroup();
54
  private CounterGroup counterGroup = new CounterGroup();
61
  public class syslogHandler extends SimpleChannelHandler {
55
  public class syslogHandler extends SimpleChannelHandler {
62
    private SyslogUtils syslogUtils = new SyslogUtils(true);
56
  private SyslogUtils syslogUtils = new SyslogUtils(maxsize, SyslogSourceConfigurationConstants.DEFAULT_KEEP_FIELDS, true);
63

    
   
57

   
64
    public void setFormater(Map<String, String> prop) {
58
    public void setFormater(Map<String, String> prop) {
65
      syslogUtils.addFormats(prop);
59
      syslogUtils.addFormats(prop);
66
    }
60
    }
67

    
   
61

   
[+20] [20] 20 lines
[+20] [+] public void messageReceived(ChannelHandlerContext ctx, MessageEvent mEvent) {
88
    // setup Netty server
82
    // setup Netty server
89
    ConnectionlessBootstrap serverBootstrap = new ConnectionlessBootstrap
83
    ConnectionlessBootstrap serverBootstrap = new ConnectionlessBootstrap
90
        (new OioDatagramChannelFactory(Executors.newCachedThreadPool()));
84
        (new OioDatagramChannelFactory(Executors.newCachedThreadPool()));
91
    final syslogHandler handler = new syslogHandler();
85
    final syslogHandler handler = new syslogHandler();
92
    handler.setFormater(formaterProp);
86
    handler.setFormater(formaterProp);

    
   
87
    serverBootstrap.setOption("receiveBufferSizePredictorFactory", new AdaptiveReceiveBufferSizePredictorFactory(maxsize, maxsize, maxsize));
93
    serverBootstrap.setPipelineFactory(new ChannelPipelineFactory() {
88
    serverBootstrap.setPipelineFactory(new ChannelPipelineFactory() {
94
      @Override
89
      @Override
95
      public ChannelPipeline getPipeline() {
90
      public ChannelPipeline getPipeline() {
96
       return Channels.pipeline(handler);
91
       return Channels.pipeline(handler);
97
      }
92
      }
[+20] [20] 40 lines
flume-ng-core/src/test/java/org/apache/flume/source/TestSyslogUdpSource.java
Revision eae26ed New Change
 
  1. flume-ng-core/src/main/java/org/apache/flume/source/SyslogUDPSource.java: Loading...
  2. flume-ng-core/src/test/java/org/apache/flume/source/TestSyslogUdpSource.java: Loading...