src/main/java/org/lwes/hadoop/io/DatagramPacketInputStream.java

   1 package org.lwes.hadoop.io;
   2
   3
   4 import java.io.DataInputStream;
   5 import java.io.EOFException;
   6 import java.io.IOException;
   7 import java.io.InputStream;
   8 import java.io.StreamCorruptedException;
   9 import java.text.SimpleDateFormat;
  10 import java.util.Date;
  11
  12 import org.lwes.serializer.DeserializerState;
  13 import org.lwes.serializer.Deserializer;
  14 import org.lwes.Event;
  15 import org.lwes.EventFactory;
  16 import org.lwes.util.IPAddress;
  17
  18 public class DatagramPacketInputStream extends DataInputStream
  19 {
  20   static int   HEADER_LENGTH    = 22;
  21   SimpleDateFormat dateFormatter = new SimpleDateFormat();
  22   private EventFactory ef      = null;
  23   private boolean typeChecking = false;  /* act as it did before */
  24
  25   public void setEventFactory(EventFactory eventFactory)
  26     { this.ef = eventFactory; }
  27   public EventFactory getEventFactory()
  28     { return ef; }
  29
  30   public void setTypeChecking(boolean typeChecking)
  31     { this.typeChecking = typeChecking; }
  32   public boolean isTypeChecking()
  33     { return typeChecking; }
  34
  35   public DatagramPacketInputStream(InputStream is)
  36     throws IOException, StreamCorruptedException
  37   {
  38     super(is);
  39   }
  40
  41   public Event readEvent()
  42   {
  43     Event event = null;
  44
  45     if ( ef == null )
  46     {
  47       ef = new EventFactory();
  48     }
  49     try {
  50       DeserializerState myState = new DeserializerState();
  51
  52       byte[] header = new byte[HEADER_LENGTH];
  53       readFully(header,0,HEADER_LENGTH);
  54       int length = (int)(Deserializer.deserializeUINT16(myState,header));
  55       long time = Deserializer.deserializeINT64(myState,header);
  56       IPAddress source_inet =
  57         new IPAddress(Deserializer.deserializeIPADDR(myState,header));
  58       int    source_port =
  59         (int)(Deserializer.deserializeUINT16(myState,header));
  60
  61       int  site_id =
  62           (int)(Deserializer.deserializeUINT16(myState,header));
  63
  64       byte [] bytes = new byte[length];
  65       readFully(bytes,0,length);
  66       if ( isTypeChecking() )
  67       {
  68         event = ef.createEvent(bytes);
  69       }
  70       else
  71       {
  72           // create event without checks
  73         event = new Event(bytes,false,null);
  74       }
  75       event.setInt64("ReceiptTime",time);
  76       event.setIPAddress("SenderIP",source_inet);
  77       event.setUInt16("SenderPort",source_port);
  78       event.setUInt16("SiteID",site_id);
  79     } catch ( EOFException e ) {
  80       event=null;
  81     } catch ( Exception e ) {
  82       System.err.println("Problem with OutputStreamFromDataGrams "+e);
  83       e.printStackTrace();
  84       event=null;
  85     }
  86     return event;
  87   }
  88
  89   public byte[] readDataGramBytes()
  90   {
  91     byte [] bytes = null;
  92     try {
  93       DeserializerState myState = new DeserializerState();
  94
  95       byte[] header = new byte[HEADER_LENGTH];
  96       readFully(header,0,HEADER_LENGTH);
  97       int length = (int)(Deserializer.deserializeUINT16(myState,header));
  98       long time = Deserializer.deserializeINT64(myState,header);
  99       IPAddress source_inet =
 100         new IPAddress(Deserializer.deserializeIPADDR(myState,header));
 101       int    source_port =
 102         (int)(Deserializer.deserializeUINT16(myState,header));
 103       int  site_id =
 104           (int)(Deserializer.deserializeUINT16(myState,header));
 105
 106       bytes = new byte[length];
 107       readFully(bytes,0,length);
 108     } catch ( EOFException e ) {
 109     } catch ( Exception e ) {
 110       System.err.println("Problem with OutputStreamFromDataGrams "+e);
 111     }
 112     return bytes;
 113   }
 114
 115   public String readDataGram()
 116   {
 117     StringBuffer sb = new StringBuffer();
 118     int total = -1;
 119     try {
 120       DeserializerState myState = new DeserializerState();
 121
 122       byte[] header = new byte[HEADER_LENGTH];
 123       total = read(header,0,HEADER_LENGTH);
 124
 125       if ( total != -1 )
 126       {
 127         int length = (int)(Deserializer.deserializeUINT16(myState,header));
 128         long time = Deserializer.deserializeINT64(myState,header);
 129         IPAddress source_inet =
 130           new IPAddress(Deserializer.deserializeIPADDR(myState,header));
 131         int    source_port =
 132           (int)(Deserializer.deserializeUINT16(myState,header));
 133         int  site_id =
 134           (int)(Deserializer.deserializeUINT16(myState,header));
 135
 136         dateFormatter.applyPattern("MM/dd/yyyy HH:mm:ss.SSS");
 137         Date date = new Date(time);
 138         sb.append(dateFormatter.format(date));
 139
 140         byte [] bytes = new byte[length];
 141         total = read(bytes,0,length);
 142       }
 143
 144     } catch ( Exception e ) {
 145       System.err.println("Problem with OutputStreamFromDataGrams "+e);
 146       e.printStackTrace();
 147     }
 148     if ( total == -1 ) return null;
 149
 150     return sb.toString();
 151   }
 152
 153 }
 154