DefaultStreamHandler implements StreamHandler { @SuppressWarnings("unchecked") @Override public void addStreamListener(final StreamListener<? extends Packet> listener) { try { final Method method = listener.getClass().getMethod("endStream", Stream.class); final ParameterizedType parameterizedType = (ParameterizedType) method.getGenericParameterTypes()[0]; final Type[] parameterArgTypes = parameterizedType.getActualTypeArguments(); final Type parameterArgType = parameterArgTypes[0]; final Class<?> parameterArgClass = (Class<?>) parameterArgType; if (parameterArgClass.equals(SipPacket.class)) { if (this.sipStreamHandler == null) { this.sipStreamHandler = new SipStreamHandler(this.framerManager); } this.sipStreamHandler.addListener((StreamListener<SipPacket>) listener); } else if (parameterArgClass.equals(RtpPacket.class)) { if (this.rtpStreamHandler == null) { this.rtpStreamHandler = new RtpStreamHandler(this.framerManager); } this.rtpStreamHandler.addListener((StreamListener<RtpPacket>) listener); } } catch (final ArrayIndexOutOfBoundsException e) { throw new RuntimeException("Unable to figure out the paramterized type", e); } catch (final SecurityException e) { throw new RuntimeException("Unable to access method information due to security constraints", e); } catch (final NoSuchMethodException e) { throw new RuntimeException("The startStream method doesn't exist. Signature changed?", e); } catch (final ClassCastException e) { throw new IllegalArgumentException("The supplied listener has not been parameterized"); } } DefaultStreamHandler(); @Override boolean nextPacket(Packet packet); @SuppressWarnings("unchecked") @Override void addStreamListener(final StreamListener<? extends Packet> listener); @Override void setFragmentListener(final FragmentListener listener); @Override SipStatistics getSipStatistics(); @Override Map<StreamId, ? extends Stream> getStreams();  }
@Test public void testRtpStream() throws Exception { final Pcap pcap = Pcap.openStream(StreamsTestBase.class.getResourceAsStream("sip_rtp.pcap")); final StreamHandler streamHandler = new DefaultStreamHandler(); final RtpCounter streamCounter = new RtpCounter(); streamHandler.addStreamListener(streamCounter); pcap.loop(streamHandler); pcap.close(); assertThat(streamCounter.startCount, is(1)); assertThat(streamCounter.packetCount, is(501)); assertThat(streamCounter.endCount, is(0)); }
@Test public void testSipStreamContainingRTPandRTCP() throws Exception { final Pcap pcap = Pcap.openStream(StreamsTestBase.class.getResourceAsStream("sip_rtp.pcap")); final StreamHandler streamHandler = new DefaultStreamHandler(); final StreamCounter streamCounter = new StreamCounter(); streamHandler.addStreamListener(streamCounter); pcap.loop(streamHandler); pcap.close(); }
@Test public void testBasicSipStreamDetection() throws Exception { final Pcap pcap = Pcap.openStream(StreamsTestBase.class.getResourceAsStream("sipp.pcap")); final StreamHandler streamHandler = new DefaultStreamHandler(); final StreamCounter streamCounter = new StreamCounter(); streamHandler.addStreamListener(streamCounter); pcap.loop(streamHandler); pcap.close(); assertThat(streamCounter.startCount, is(5)); assertThat(streamCounter.packetCount, is(30)); assertThat(streamCounter.endCount, is(5)); assertStream(streamCounter.streams.get(0), 408, 1002719); assertStream(streamCounter.streams.get(1), 400, 1003484); assertStream(streamCounter.streams.get(2), 315, 1002358); assertStream(streamCounter.streams.get(3), 369, 1001320); assertStream(streamCounter.streams.get(4), 470, 1002963); }
@SuppressWarnings({ "unchecked", "rawtypes" }) @Test(expected = IllegalArgumentException.class) public void testNotParamterizedStreamListener() throws Exception { final StreamHandler streamHandler = new DefaultStreamHandler(); streamHandler.addStreamListener(new StreamListener() { @Override public void startStream(final Stream stream, final Packet packet) { } @Override public void packetReceived(final Stream stream, final Packet packet) { } @Override public void endStream(final Stream stream) { } }); }