@Test public void testNullFromGetTransaction() { Channel ch = mock(Channel.class); when(ch.getTransaction()).thenReturn(null); ChannelSelector sel = new ReplicatingChannelSelector(); sel.setChannels(Lists.newArrayList(ch)); ChannelProcessor proc = new ChannelProcessor(sel); List<Event> events = Lists.newArrayList(); events.add(EventBuilder.withBody("event 1", Charsets.UTF_8)); boolean threw = false; try { proc.processEventBatch(events); } catch (NullPointerException ex) { threw = true; Assert.assertNotNull("NPE must be manually thrown", ex.getMessage()); } Assert.assertTrue("Must throw NPE", threw); }
public void processEventBatch(List<Event> events) { Preconditions.checkNotNull(events, "Event list must not be null"); events = interceptorChain.intercept(events); Map<Channel, List<Event>> reqChannelQueue = new LinkedHashMap<Channel, List<Event>>(); Map<Channel, List<Event>> optChannelQueue = new LinkedHashMap<Channel, List<Event>>(); for (Event event : events) { List<Channel> reqChannels = selector.getRequiredChannels(event); for (Channel ch : reqChannels) { List<Event> eventQueue = reqChannelQueue.get(ch); if (eventQueue == null) { eventQueue = new ArrayList<Event>(); reqChannelQueue.put(ch, eventQueue); } eventQueue.add(event); } List<Channel> optChannels = selector.getOptionalChannels(event); for (Channel ch: optChannels) { List<Event> eventQueue = optChannelQueue.get(ch); if (eventQueue == null) { eventQueue = new ArrayList<Event>(); optChannelQueue.put(ch, eventQueue); } eventQueue.add(event); } } for (Channel reqChannel : reqChannelQueue.keySet()) { Transaction tx = reqChannel.getTransaction(); Preconditions.checkNotNull(tx, "Transaction object must not be null"); try { tx.begin(); List<Event> batch = reqChannelQueue.get(reqChannel); for (Event event : batch) { reqChannel.put(event); } tx.commit(); } catch (Throwable t) { tx.rollback(); if (t instanceof Error) { LOG.error("Error while writing to required channel: " + reqChannel, t); throw (Error) t; } else { throw new ChannelException("Unable to put batch on required " + "channel: " + reqChannel, t); } } finally { if (tx != null) { tx.close(); } } } for (Channel optChannel : optChannelQueue.keySet()) { Transaction tx = optChannel.getTransaction(); Preconditions.checkNotNull(tx, "Transaction object must not be null"); try { tx.begin(); List<Event> batch = optChannelQueue.get(optChannel); for (Event event : batch ) { optChannel.put(event); } tx.commit(); } catch (Throwable t) { tx.rollback(); LOG.error("Unable to put batch on optional channel: " + optChannel, t); if (t instanceof Error) { throw (Error) t; } } finally { if (tx != null) { tx.close(); } } } }
ChannelProcessor implements Configurable { public void processEventBatch(List<Event> events) { Preconditions.checkNotNull(events, "Event list must not be null"); events = interceptorChain.intercept(events); Map<Channel, List<Event>> reqChannelQueue = new LinkedHashMap<Channel, List<Event>>(); Map<Channel, List<Event>> optChannelQueue = new LinkedHashMap<Channel, List<Event>>(); for (Event event : events) { List<Channel> reqChannels = selector.getRequiredChannels(event); for (Channel ch : reqChannels) { List<Event> eventQueue = reqChannelQueue.get(ch); if (eventQueue == null) { eventQueue = new ArrayList<Event>(); reqChannelQueue.put(ch, eventQueue); } eventQueue.add(event); } List<Channel> optChannels = selector.getOptionalChannels(event); for (Channel ch: optChannels) { List<Event> eventQueue = optChannelQueue.get(ch); if (eventQueue == null) { eventQueue = new ArrayList<Event>(); optChannelQueue.put(ch, eventQueue); } eventQueue.add(event); } } for (Channel reqChannel : reqChannelQueue.keySet()) { Transaction tx = reqChannel.getTransaction(); Preconditions.checkNotNull(tx, "Transaction object must not be null"); try { tx.begin(); List<Event> batch = reqChannelQueue.get(reqChannel); for (Event event : batch) { reqChannel.put(event); } tx.commit(); } catch (Throwable t) { tx.rollback(); if (t instanceof Error) { LOG.error("Error while writing to required channel: " + reqChannel, t); throw (Error) t; } else { throw new ChannelException("Unable to put batch on required " + "channel: " + reqChannel, t); } } finally { if (tx != null) { tx.close(); } } } for (Channel optChannel : optChannelQueue.keySet()) { Transaction tx = optChannel.getTransaction(); Preconditions.checkNotNull(tx, "Transaction object must not be null"); try { tx.begin(); List<Event> batch = optChannelQueue.get(optChannel); for (Event event : batch ) { optChannel.put(event); } tx.commit(); } catch (Throwable t) { tx.rollback(); LOG.error("Unable to put batch on optional channel: " + optChannel, t); if (t instanceof Error) { throw (Error) t; } } finally { if (tx != null) { tx.close(); } } } } }
ChannelProcessor implements Configurable { public void processEventBatch(List<Event> events) { Preconditions.checkNotNull(events, "Event list must not be null"); events = interceptorChain.intercept(events); Map<Channel, List<Event>> reqChannelQueue = new LinkedHashMap<Channel, List<Event>>(); Map<Channel, List<Event>> optChannelQueue = new LinkedHashMap<Channel, List<Event>>(); for (Event event : events) { List<Channel> reqChannels = selector.getRequiredChannels(event); for (Channel ch : reqChannels) { List<Event> eventQueue = reqChannelQueue.get(ch); if (eventQueue == null) { eventQueue = new ArrayList<Event>(); reqChannelQueue.put(ch, eventQueue); } eventQueue.add(event); } List<Channel> optChannels = selector.getOptionalChannels(event); for (Channel ch: optChannels) { List<Event> eventQueue = optChannelQueue.get(ch); if (eventQueue == null) { eventQueue = new ArrayList<Event>(); optChannelQueue.put(ch, eventQueue); } eventQueue.add(event); } } for (Channel reqChannel : reqChannelQueue.keySet()) { Transaction tx = reqChannel.getTransaction(); Preconditions.checkNotNull(tx, "Transaction object must not be null"); try { tx.begin(); List<Event> batch = reqChannelQueue.get(reqChannel); for (Event event : batch) { reqChannel.put(event); } tx.commit(); } catch (Throwable t) { tx.rollback(); if (t instanceof Error) { LOG.error("Error while writing to required channel: " + reqChannel, t); throw (Error) t; } else { throw new ChannelException("Unable to put batch on required " + "channel: " + reqChannel, t); } } finally { if (tx != null) { tx.close(); } } } for (Channel optChannel : optChannelQueue.keySet()) { Transaction tx = optChannel.getTransaction(); Preconditions.checkNotNull(tx, "Transaction object must not be null"); try { tx.begin(); List<Event> batch = optChannelQueue.get(optChannel); for (Event event : batch ) { optChannel.put(event); } tx.commit(); } catch (Throwable t) { tx.rollback(); LOG.error("Unable to put batch on optional channel: " + optChannel, t); if (t instanceof Error) { throw (Error) t; } } finally { if (tx != null) { tx.close(); } } } } ChannelProcessor(ChannelSelector selector); }
ChannelProcessor implements Configurable { public void processEventBatch(List<Event> events) { Preconditions.checkNotNull(events, "Event list must not be null"); events = interceptorChain.intercept(events); Map<Channel, List<Event>> reqChannelQueue = new LinkedHashMap<Channel, List<Event>>(); Map<Channel, List<Event>> optChannelQueue = new LinkedHashMap<Channel, List<Event>>(); for (Event event : events) { List<Channel> reqChannels = selector.getRequiredChannels(event); for (Channel ch : reqChannels) { List<Event> eventQueue = reqChannelQueue.get(ch); if (eventQueue == null) { eventQueue = new ArrayList<Event>(); reqChannelQueue.put(ch, eventQueue); } eventQueue.add(event); } List<Channel> optChannels = selector.getOptionalChannels(event); for (Channel ch: optChannels) { List<Event> eventQueue = optChannelQueue.get(ch); if (eventQueue == null) { eventQueue = new ArrayList<Event>(); optChannelQueue.put(ch, eventQueue); } eventQueue.add(event); } } for (Channel reqChannel : reqChannelQueue.keySet()) { Transaction tx = reqChannel.getTransaction(); Preconditions.checkNotNull(tx, "Transaction object must not be null"); try { tx.begin(); List<Event> batch = reqChannelQueue.get(reqChannel); for (Event event : batch) { reqChannel.put(event); } tx.commit(); } catch (Throwable t) { tx.rollback(); if (t instanceof Error) { LOG.error("Error while writing to required channel: " + reqChannel, t); throw (Error) t; } else { throw new ChannelException("Unable to put batch on required " + "channel: " + reqChannel, t); } } finally { if (tx != null) { tx.close(); } } } for (Channel optChannel : optChannelQueue.keySet()) { Transaction tx = optChannel.getTransaction(); Preconditions.checkNotNull(tx, "Transaction object must not be null"); try { tx.begin(); List<Event> batch = optChannelQueue.get(optChannel); for (Event event : batch ) { optChannel.put(event); } tx.commit(); } catch (Throwable t) { tx.rollback(); LOG.error("Unable to put batch on optional channel: " + optChannel, t); if (t instanceof Error) { throw (Error) t; } } finally { if (tx != null) { tx.close(); } } } } ChannelProcessor(ChannelSelector selector); void initialize(); void close(); @Override void configure(Context context); ChannelSelector getSelector(); void processEventBatch(List<Event> events); void processEvent(Event event); }
ChannelProcessor implements Configurable { public void processEventBatch(List<Event> events) { Preconditions.checkNotNull(events, "Event list must not be null"); events = interceptorChain.intercept(events); Map<Channel, List<Event>> reqChannelQueue = new LinkedHashMap<Channel, List<Event>>(); Map<Channel, List<Event>> optChannelQueue = new LinkedHashMap<Channel, List<Event>>(); for (Event event : events) { List<Channel> reqChannels = selector.getRequiredChannels(event); for (Channel ch : reqChannels) { List<Event> eventQueue = reqChannelQueue.get(ch); if (eventQueue == null) { eventQueue = new ArrayList<Event>(); reqChannelQueue.put(ch, eventQueue); } eventQueue.add(event); } List<Channel> optChannels = selector.getOptionalChannels(event); for (Channel ch: optChannels) { List<Event> eventQueue = optChannelQueue.get(ch); if (eventQueue == null) { eventQueue = new ArrayList<Event>(); optChannelQueue.put(ch, eventQueue); } eventQueue.add(event); } } for (Channel reqChannel : reqChannelQueue.keySet()) { Transaction tx = reqChannel.getTransaction(); Preconditions.checkNotNull(tx, "Transaction object must not be null"); try { tx.begin(); List<Event> batch = reqChannelQueue.get(reqChannel); for (Event event : batch) { reqChannel.put(event); } tx.commit(); } catch (Throwable t) { tx.rollback(); if (t instanceof Error) { LOG.error("Error while writing to required channel: " + reqChannel, t); throw (Error) t; } else { throw new ChannelException("Unable to put batch on required " + "channel: " + reqChannel, t); } } finally { if (tx != null) { tx.close(); } } } for (Channel optChannel : optChannelQueue.keySet()) { Transaction tx = optChannel.getTransaction(); Preconditions.checkNotNull(tx, "Transaction object must not be null"); try { tx.begin(); List<Event> batch = optChannelQueue.get(optChannel); for (Event event : batch ) { optChannel.put(event); } tx.commit(); } catch (Throwable t) { tx.rollback(); LOG.error("Unable to put batch on optional channel: " + optChannel, t); if (t instanceof Error) { throw (Error) t; } } finally { if (tx != null) { tx.close(); } } } } ChannelProcessor(ChannelSelector selector); void initialize(); void close(); @Override void configure(Context context); ChannelSelector getSelector(); void processEventBatch(List<Event> events); void processEvent(Event event);  }