@Test(expected = IllegalArgumentException.class) public void nullEvent() throws RedisSerializerException { LogstashDeSerializer logstashDeSerializer = new LogstashDeSerializer(); logstashDeSerializer.parseEvent(null); }
@Override public Event parseEvent(byte[] eventSource) throws RedisSerializerException { Event flumeEvent = new SimpleEvent(); if (eventSource == null) { throw new IllegalArgumentException("Event source cannot be null"); } try { LogstashEvent logstashEvent = objectMapper.readValue(eventSource, LogstashEvent.class); Map<String, String> headers = new HashMap<String, String>(); if (StringUtils.isNotBlank(logstashEvent.getMessage())) { headers.put(FIELD_MESSAGE, logstashEvent.getMessage()); } if (logstashEvent.getFields() != null) { headers.putAll(logstashEvent.getFields()); } if (StringUtils.isNotBlank(logstashEvent.getSource())) { headers.put(FIELD_AT_SOURCE, logstashEvent.getSource()); } if (StringUtils.isNotBlank(logstashEvent.getSourceHost())) { headers.put(FIELD_AT_SOURCE_HOST, logstashEvent.getSourceHost()); } if (StringUtils.isNotBlank(logstashEvent.getSourcePath())) { headers.put(FIELD_AT_SOURCE_PATH, logstashEvent.getSourcePath()); } if (StringUtils.isNotBlank(logstashEvent.getType())) { headers.put(FIELD_AT_TYPE, logstashEvent.getType()); } if (logstashEvent.getVersion() != null) { headers.put(FIELD_AT_VERSION, logstashEvent.getVersion().toString()); } if (logstashEvent.getTimestamp() != null) { DateFormat dateFormat = new SimpleDateFormat(TIMESTAMP_FORMAT); Date timestamp = logstashEvent.getTimestamp(); headers.put(FIELD_AT_TIMESTAMP, dateFormat.format(timestamp)); } if(CollectionUtils.isNotEmpty(logstashEvent.getTags())) { headers.put(FIELD_TAGS, StringUtils.join(logstashEvent.getTags(), ",")); } flumeEvent = EventBuilder.withBody(eventSource, headers); } catch (JsonParseException e) { throw new RedisSerializerException("Could not parse event: [" + new String(eventSource) + "]", e); } catch (JsonMappingException e) { throw new RedisSerializerException("Could not parse event: [" + new String(eventSource) + "]", e); } catch (IOException e) { throw new RedisSerializerException("Could not parse event: [" + new String(eventSource) + "]", e); } return flumeEvent; }
LogstashDeSerializer implements Serializer { @Override public Event parseEvent(byte[] eventSource) throws RedisSerializerException { Event flumeEvent = new SimpleEvent(); if (eventSource == null) { throw new IllegalArgumentException("Event source cannot be null"); } try { LogstashEvent logstashEvent = objectMapper.readValue(eventSource, LogstashEvent.class); Map<String, String> headers = new HashMap<String, String>(); if (StringUtils.isNotBlank(logstashEvent.getMessage())) { headers.put(FIELD_MESSAGE, logstashEvent.getMessage()); } if (logstashEvent.getFields() != null) { headers.putAll(logstashEvent.getFields()); } if (StringUtils.isNotBlank(logstashEvent.getSource())) { headers.put(FIELD_AT_SOURCE, logstashEvent.getSource()); } if (StringUtils.isNotBlank(logstashEvent.getSourceHost())) { headers.put(FIELD_AT_SOURCE_HOST, logstashEvent.getSourceHost()); } if (StringUtils.isNotBlank(logstashEvent.getSourcePath())) { headers.put(FIELD_AT_SOURCE_PATH, logstashEvent.getSourcePath()); } if (StringUtils.isNotBlank(logstashEvent.getType())) { headers.put(FIELD_AT_TYPE, logstashEvent.getType()); } if (logstashEvent.getVersion() != null) { headers.put(FIELD_AT_VERSION, logstashEvent.getVersion().toString()); } if (logstashEvent.getTimestamp() != null) { DateFormat dateFormat = new SimpleDateFormat(TIMESTAMP_FORMAT); Date timestamp = logstashEvent.getTimestamp(); headers.put(FIELD_AT_TIMESTAMP, dateFormat.format(timestamp)); } if(CollectionUtils.isNotEmpty(logstashEvent.getTags())) { headers.put(FIELD_TAGS, StringUtils.join(logstashEvent.getTags(), ",")); } flumeEvent = EventBuilder.withBody(eventSource, headers); } catch (JsonParseException e) { throw new RedisSerializerException("Could not parse event: [" + new String(eventSource) + "]", e); } catch (JsonMappingException e) { throw new RedisSerializerException("Could not parse event: [" + new String(eventSource) + "]", e); } catch (IOException e) { throw new RedisSerializerException("Could not parse event: [" + new String(eventSource) + "]", e); } return flumeEvent; } }
LogstashDeSerializer implements Serializer { @Override public Event parseEvent(byte[] eventSource) throws RedisSerializerException { Event flumeEvent = new SimpleEvent(); if (eventSource == null) { throw new IllegalArgumentException("Event source cannot be null"); } try { LogstashEvent logstashEvent = objectMapper.readValue(eventSource, LogstashEvent.class); Map<String, String> headers = new HashMap<String, String>(); if (StringUtils.isNotBlank(logstashEvent.getMessage())) { headers.put(FIELD_MESSAGE, logstashEvent.getMessage()); } if (logstashEvent.getFields() != null) { headers.putAll(logstashEvent.getFields()); } if (StringUtils.isNotBlank(logstashEvent.getSource())) { headers.put(FIELD_AT_SOURCE, logstashEvent.getSource()); } if (StringUtils.isNotBlank(logstashEvent.getSourceHost())) { headers.put(FIELD_AT_SOURCE_HOST, logstashEvent.getSourceHost()); } if (StringUtils.isNotBlank(logstashEvent.getSourcePath())) { headers.put(FIELD_AT_SOURCE_PATH, logstashEvent.getSourcePath()); } if (StringUtils.isNotBlank(logstashEvent.getType())) { headers.put(FIELD_AT_TYPE, logstashEvent.getType()); } if (logstashEvent.getVersion() != null) { headers.put(FIELD_AT_VERSION, logstashEvent.getVersion().toString()); } if (logstashEvent.getTimestamp() != null) { DateFormat dateFormat = new SimpleDateFormat(TIMESTAMP_FORMAT); Date timestamp = logstashEvent.getTimestamp(); headers.put(FIELD_AT_TIMESTAMP, dateFormat.format(timestamp)); } if(CollectionUtils.isNotEmpty(logstashEvent.getTags())) { headers.put(FIELD_TAGS, StringUtils.join(logstashEvent.getTags(), ",")); } flumeEvent = EventBuilder.withBody(eventSource, headers); } catch (JsonParseException e) { throw new RedisSerializerException("Could not parse event: [" + new String(eventSource) + "]", e); } catch (JsonMappingException e) { throw new RedisSerializerException("Could not parse event: [" + new String(eventSource) + "]", e); } catch (IOException e) { throw new RedisSerializerException("Could not parse event: [" + new String(eventSource) + "]", e); } return flumeEvent; }  }
LogstashDeSerializer implements Serializer { @Override public Event parseEvent(byte[] eventSource) throws RedisSerializerException { Event flumeEvent = new SimpleEvent(); if (eventSource == null) { throw new IllegalArgumentException("Event source cannot be null"); } try { LogstashEvent logstashEvent = objectMapper.readValue(eventSource, LogstashEvent.class); Map<String, String> headers = new HashMap<String, String>(); if (StringUtils.isNotBlank(logstashEvent.getMessage())) { headers.put(FIELD_MESSAGE, logstashEvent.getMessage()); } if (logstashEvent.getFields() != null) { headers.putAll(logstashEvent.getFields()); } if (StringUtils.isNotBlank(logstashEvent.getSource())) { headers.put(FIELD_AT_SOURCE, logstashEvent.getSource()); } if (StringUtils.isNotBlank(logstashEvent.getSourceHost())) { headers.put(FIELD_AT_SOURCE_HOST, logstashEvent.getSourceHost()); } if (StringUtils.isNotBlank(logstashEvent.getSourcePath())) { headers.put(FIELD_AT_SOURCE_PATH, logstashEvent.getSourcePath()); } if (StringUtils.isNotBlank(logstashEvent.getType())) { headers.put(FIELD_AT_TYPE, logstashEvent.getType()); } if (logstashEvent.getVersion() != null) { headers.put(FIELD_AT_VERSION, logstashEvent.getVersion().toString()); } if (logstashEvent.getTimestamp() != null) { DateFormat dateFormat = new SimpleDateFormat(TIMESTAMP_FORMAT); Date timestamp = logstashEvent.getTimestamp(); headers.put(FIELD_AT_TIMESTAMP, dateFormat.format(timestamp)); } if(CollectionUtils.isNotEmpty(logstashEvent.getTags())) { headers.put(FIELD_TAGS, StringUtils.join(logstashEvent.getTags(), ",")); } flumeEvent = EventBuilder.withBody(eventSource, headers); } catch (JsonParseException e) { throw new RedisSerializerException("Could not parse event: [" + new String(eventSource) + "]", e); } catch (JsonMappingException e) { throw new RedisSerializerException("Could not parse event: [" + new String(eventSource) + "]", e); } catch (IOException e) { throw new RedisSerializerException("Could not parse event: [" + new String(eventSource) + "]", e); } return flumeEvent; }  @Override void configure(Context context); @Override void configure(ComponentConfiguration conf); @Override Event parseEvent(byte[] eventSource); }
LogstashDeSerializer implements Serializer { @Override public Event parseEvent(byte[] eventSource) throws RedisSerializerException { Event flumeEvent = new SimpleEvent(); if (eventSource == null) { throw new IllegalArgumentException("Event source cannot be null"); } try { LogstashEvent logstashEvent = objectMapper.readValue(eventSource, LogstashEvent.class); Map<String, String> headers = new HashMap<String, String>(); if (StringUtils.isNotBlank(logstashEvent.getMessage())) { headers.put(FIELD_MESSAGE, logstashEvent.getMessage()); } if (logstashEvent.getFields() != null) { headers.putAll(logstashEvent.getFields()); } if (StringUtils.isNotBlank(logstashEvent.getSource())) { headers.put(FIELD_AT_SOURCE, logstashEvent.getSource()); } if (StringUtils.isNotBlank(logstashEvent.getSourceHost())) { headers.put(FIELD_AT_SOURCE_HOST, logstashEvent.getSourceHost()); } if (StringUtils.isNotBlank(logstashEvent.getSourcePath())) { headers.put(FIELD_AT_SOURCE_PATH, logstashEvent.getSourcePath()); } if (StringUtils.isNotBlank(logstashEvent.getType())) { headers.put(FIELD_AT_TYPE, logstashEvent.getType()); } if (logstashEvent.getVersion() != null) { headers.put(FIELD_AT_VERSION, logstashEvent.getVersion().toString()); } if (logstashEvent.getTimestamp() != null) { DateFormat dateFormat = new SimpleDateFormat(TIMESTAMP_FORMAT); Date timestamp = logstashEvent.getTimestamp(); headers.put(FIELD_AT_TIMESTAMP, dateFormat.format(timestamp)); } if(CollectionUtils.isNotEmpty(logstashEvent.getTags())) { headers.put(FIELD_TAGS, StringUtils.join(logstashEvent.getTags(), ",")); } flumeEvent = EventBuilder.withBody(eventSource, headers); } catch (JsonParseException e) { throw new RedisSerializerException("Could not parse event: [" + new String(eventSource) + "]", e); } catch (JsonMappingException e) { throw new RedisSerializerException("Could not parse event: [" + new String(eventSource) + "]", e); } catch (IOException e) { throw new RedisSerializerException("Could not parse event: [" + new String(eventSource) + "]", e); } return flumeEvent; }  @Override void configure(Context context); @Override void configure(ComponentConfiguration conf); @Override Event parseEvent(byte[] eventSource); static final String TIMESTAMP_FORMAT; static final String FIELD_AT_TIMESTAMP; static final String FIELD_AT_TYPE; static final String FIELD_AT_SOURCE_PATH; static final String FIELD_AT_SOURCE_HOST; static final String FIELD_AT_SOURCE; static final String FIELD_AT_VERSION; static final String FIELD_TAGS; static final String FIELD_MESSAGE; }