### Question:
YoutubeUserActivityCollector extends YoutubeDataCollector { void processActivityFeed(ActivityListResponse feed, DateTime afterDate, DateTime beforeDate) throws IOException, InterruptedException { for (com.google.api.services.youtube.model.Activity activity : feed.getItems()) { try { List<Video> videos = new ArrayList<>(); if (activity.getContentDetails().getUpload() != null) { videos.addAll(getVideoList(activity.getContentDetails().getUpload().getVideoId())); } if (activity.getContentDetails().getPlaylistItem() != null && activity.getContentDetails().getPlaylistItem().getResourceId() != null) { videos.addAll(getVideoList(activity.getContentDetails().getPlaylistItem().getResourceId().getVideoId())); } processVideos(videos, afterDate, beforeDate, activity, feed); } catch (Exception ex) { LOGGER.error("Error while trying to process activity: {}, {}", activity, ex); } } } YoutubeUserActivityCollector(
      YouTube youtube,
      BlockingQueue<StreamsDatum> datumQueue,
      BackOffStrategy backOff,
      UserInfo userInfo,
      YoutubeConfiguration config); @Override void run();  }### Answer:
@Test public void testProcessActivityFeed() throws IOException, InterruptedException { DateTime now = new DateTime(System.currentTimeMillis()); YouTube youtube = buildYouTube(0, 0, 5, now.plus(3000000), now.minus(1000000)); BlockingQueue<StreamsDatum> datumQueue = new ThroughputQueue<>(); YoutubeUserActivityCollector collector = new YoutubeUserActivityCollector(youtube, datumQueue, new ExponentialBackOffStrategy(2), new UserInfo().withUserId(USER_ID), this.config); ActivityListResponse feed = buildActivityListResponse(1); collector.processActivityFeed(feed, new DateTime(System.currentTimeMillis()), null); assertEquals(collector.getDatumQueue().size(), 5); }