@Test public void testObjectMessage() throws Exception { createObjectMessage(); headers.put("key1", "value1"); headers.put("key2", "value2"); createHeaders(); Event event = converter.convert(message).iterator().next(); assertEquals(headers, event.getHeaders()); ByteArrayOutputStream bos = new ByteArrayOutputStream(); ObjectOutput out = new ObjectOutputStream(bos); out.writeObject(TEXT); assertArrayEquals(bos.toByteArray(), event.getBody()); }
@Override public List<Event> convert(Message message) throws JMSException { Event event = new SimpleEvent(); Map<String, String> headers = event.getHeaders(); @SuppressWarnings("rawtypes") Enumeration propertyNames = message.getPropertyNames(); while(propertyNames.hasMoreElements()) { String name = propertyNames.nextElement().toString(); String value = message.getStringProperty(name); headers.put(name, value); } if(message instanceof BytesMessage) { BytesMessage bytesMessage = (BytesMessage)message; long length = bytesMessage.getBodyLength(); if(length > 0L) { if (length > Integer.MAX_VALUE) { throw new JMSException("Unable to process message " + "of size " + length); } byte[] body = new byte[(int)length]; int count = bytesMessage.readBytes(body); if(count != length) { throw new JMSException("Unable to read full message. " + "Read " + count + " of total " + length); } event.setBody(body); } } else if(message instanceof TextMessage) { TextMessage textMessage = (TextMessage)message; event.setBody(textMessage.getText().getBytes(charset)); } else if(message instanceof ObjectMessage) { ObjectMessage objectMessage = (ObjectMessage)message; Object object = objectMessage.getObject(); if(object != null) { ByteArrayOutputStream bos = new ByteArrayOutputStream(); ObjectOutput out = null; try { out = new ObjectOutputStream(bos); out.writeObject(object); event.setBody(bos.toByteArray()); } catch(IOException e) { throw new FlumeException("Error serializing object", e); } finally { try { if(out != null) { out.close(); } } catch (IOException e) { throw new FlumeException("Error closing ObjectOutputStream", e); } try { if(bos != null) { bos.close(); } } catch (IOException e) { throw new FlumeException("Error closing ByteArrayOutputStream", e); } } } } List<Event> events = new ArrayList<Event>(1); events.add(event); return events; }
DefaultJMSMessageConverter implements JMSMessageConverter { @Override public List<Event> convert(Message message) throws JMSException { Event event = new SimpleEvent(); Map<String, String> headers = event.getHeaders(); @SuppressWarnings("rawtypes") Enumeration propertyNames = message.getPropertyNames(); while(propertyNames.hasMoreElements()) { String name = propertyNames.nextElement().toString(); String value = message.getStringProperty(name); headers.put(name, value); } if(message instanceof BytesMessage) { BytesMessage bytesMessage = (BytesMessage)message; long length = bytesMessage.getBodyLength(); if(length > 0L) { if (length > Integer.MAX_VALUE) { throw new JMSException("Unable to process message " + "of size " + length); } byte[] body = new byte[(int)length]; int count = bytesMessage.readBytes(body); if(count != length) { throw new JMSException("Unable to read full message. " + "Read " + count + " of total " + length); } event.setBody(body); } } else if(message instanceof TextMessage) { TextMessage textMessage = (TextMessage)message; event.setBody(textMessage.getText().getBytes(charset)); } else if(message instanceof ObjectMessage) { ObjectMessage objectMessage = (ObjectMessage)message; Object object = objectMessage.getObject(); if(object != null) { ByteArrayOutputStream bos = new ByteArrayOutputStream(); ObjectOutput out = null; try { out = new ObjectOutputStream(bos); out.writeObject(object); event.setBody(bos.toByteArray()); } catch(IOException e) { throw new FlumeException("Error serializing object", e); } finally { try { if(out != null) { out.close(); } } catch (IOException e) { throw new FlumeException("Error closing ObjectOutputStream", e); } try { if(bos != null) { bos.close(); } } catch (IOException e) { throw new FlumeException("Error closing ByteArrayOutputStream", e); } } } } List<Event> events = new ArrayList<Event>(1); events.add(event); return events; } }
DefaultJMSMessageConverter implements JMSMessageConverter { @Override public List<Event> convert(Message message) throws JMSException { Event event = new SimpleEvent(); Map<String, String> headers = event.getHeaders(); @SuppressWarnings("rawtypes") Enumeration propertyNames = message.getPropertyNames(); while(propertyNames.hasMoreElements()) { String name = propertyNames.nextElement().toString(); String value = message.getStringProperty(name); headers.put(name, value); } if(message instanceof BytesMessage) { BytesMessage bytesMessage = (BytesMessage)message; long length = bytesMessage.getBodyLength(); if(length > 0L) { if (length > Integer.MAX_VALUE) { throw new JMSException("Unable to process message " + "of size " + length); } byte[] body = new byte[(int)length]; int count = bytesMessage.readBytes(body); if(count != length) { throw new JMSException("Unable to read full message. " + "Read " + count + " of total " + length); } event.setBody(body); } } else if(message instanceof TextMessage) { TextMessage textMessage = (TextMessage)message; event.setBody(textMessage.getText().getBytes(charset)); } else if(message instanceof ObjectMessage) { ObjectMessage objectMessage = (ObjectMessage)message; Object object = objectMessage.getObject(); if(object != null) { ByteArrayOutputStream bos = new ByteArrayOutputStream(); ObjectOutput out = null; try { out = new ObjectOutputStream(bos); out.writeObject(object); event.setBody(bos.toByteArray()); } catch(IOException e) { throw new FlumeException("Error serializing object", e); } finally { try { if(out != null) { out.close(); } } catch (IOException e) { throw new FlumeException("Error closing ObjectOutputStream", e); } try { if(bos != null) { bos.close(); } } catch (IOException e) { throw new FlumeException("Error closing ByteArrayOutputStream", e); } } } } List<Event> events = new ArrayList<Event>(1); events.add(event); return events; } private  DefaultJMSMessageConverter(String charset); }
DefaultJMSMessageConverter implements JMSMessageConverter { @Override public List<Event> convert(Message message) throws JMSException { Event event = new SimpleEvent(); Map<String, String> headers = event.getHeaders(); @SuppressWarnings("rawtypes") Enumeration propertyNames = message.getPropertyNames(); while(propertyNames.hasMoreElements()) { String name = propertyNames.nextElement().toString(); String value = message.getStringProperty(name); headers.put(name, value); } if(message instanceof BytesMessage) { BytesMessage bytesMessage = (BytesMessage)message; long length = bytesMessage.getBodyLength(); if(length > 0L) { if (length > Integer.MAX_VALUE) { throw new JMSException("Unable to process message " + "of size " + length); } byte[] body = new byte[(int)length]; int count = bytesMessage.readBytes(body); if(count != length) { throw new JMSException("Unable to read full message. " + "Read " + count + " of total " + length); } event.setBody(body); } } else if(message instanceof TextMessage) { TextMessage textMessage = (TextMessage)message; event.setBody(textMessage.getText().getBytes(charset)); } else if(message instanceof ObjectMessage) { ObjectMessage objectMessage = (ObjectMessage)message; Object object = objectMessage.getObject(); if(object != null) { ByteArrayOutputStream bos = new ByteArrayOutputStream(); ObjectOutput out = null; try { out = new ObjectOutputStream(bos); out.writeObject(object); event.setBody(bos.toByteArray()); } catch(IOException e) { throw new FlumeException("Error serializing object", e); } finally { try { if(out != null) { out.close(); } } catch (IOException e) { throw new FlumeException("Error closing ObjectOutputStream", e); } try { if(bos != null) { bos.close(); } } catch (IOException e) { throw new FlumeException("Error closing ByteArrayOutputStream", e); } } } } List<Event> events = new ArrayList<Event>(1); events.add(event); return events; } private  DefaultJMSMessageConverter(String charset); @Override List<Event> convert(Message message); }
DefaultJMSMessageConverter implements JMSMessageConverter { @Override public List<Event> convert(Message message) throws JMSException { Event event = new SimpleEvent(); Map<String, String> headers = event.getHeaders(); @SuppressWarnings("rawtypes") Enumeration propertyNames = message.getPropertyNames(); while(propertyNames.hasMoreElements()) { String name = propertyNames.nextElement().toString(); String value = message.getStringProperty(name); headers.put(name, value); } if(message instanceof BytesMessage) { BytesMessage bytesMessage = (BytesMessage)message; long length = bytesMessage.getBodyLength(); if(length > 0L) { if (length > Integer.MAX_VALUE) { throw new JMSException("Unable to process message " + "of size " + length); } byte[] body = new byte[(int)length]; int count = bytesMessage.readBytes(body); if(count != length) { throw new JMSException("Unable to read full message. " + "Read " + count + " of total " + length); } event.setBody(body); } } else if(message instanceof TextMessage) { TextMessage textMessage = (TextMessage)message; event.setBody(textMessage.getText().getBytes(charset)); } else if(message instanceof ObjectMessage) { ObjectMessage objectMessage = (ObjectMessage)message; Object object = objectMessage.getObject(); if(object != null) { ByteArrayOutputStream bos = new ByteArrayOutputStream(); ObjectOutput out = null; try { out = new ObjectOutputStream(bos); out.writeObject(object); event.setBody(bos.toByteArray()); } catch(IOException e) { throw new FlumeException("Error serializing object", e); } finally { try { if(out != null) { out.close(); } } catch (IOException e) { throw new FlumeException("Error closing ObjectOutputStream", e); } try { if(bos != null) { bos.close(); } } catch (IOException e) { throw new FlumeException("Error closing ByteArrayOutputStream", e); } } } } List<Event> events = new ArrayList<Event>(1); events.add(event); return events; } private  DefaultJMSMessageConverter(String charset); @Override List<Event> convert(Message message);  }