code: public static <T> Source<T> filter(final Source<T> source,
                                         final Filter<T> filter) {
          return new AbstractSource<T>() {
              @Override
              public T computeNext() throws IOException {
                  while (source.hasNext()) {
                      T item = filter.accept(source.next());
                      if (item != null) {
                          return item;
                      }
                  }
                  return endOfData();
              }
  
              @Override
              public void close() {
                  source.close();
              }
          };
      }
docs: Gets a version of the given source whose items are subject to the
  given filter. Items may be omitted or transformed by the filter.
  <p>
  When the returned source is closed, the wrapped source will also be
  closed, but the filter will not be.
  
  @param source the source to filter.
  @param filter the filter to use.
  @param <T> the type.
  @return the filtering source.