twitter4j - получить доступ к информации о твите из Streaming API

Моя цель - собрать все твиты, содержащие слова «Франция» и «Германия», а также собрать связанные метаданные (например, географические координаты, прикрепленные к е твитнуть). Я знаю, что эти метаданные доступны, но я не могу понять, как получить к ним доступ с помощью библиотеки Java, которую я использую: «twitter4j».

Итак, то, что у меня есть, взято из примеров кода на сайте twitter4j. Он распечатывает все твиты, содержащие выбранные мной ключевые слова, поскольку они предоставляются в реальном времени с помощью Streaming API Twitter. Я вызываю метод фильтра в своем объекте TwitterStream, и он предоставляет поток. Но мне нужно больше контроля. А именно, я хотел бы иметь возможность:

1) записывать твиты в файл; 2) распечатать только первые 1000 твитов; 3) получить доступ к другим метаданным, прикрепленным к твиту (метод фильтрации просто выводит имя пользователя и сам твит).

Вот код, который у меня есть:

import twitter4j.FilterQuery;
import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;
import twitter4j.TwitterException;
import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.conf.ConfigurationBuilder;

public class Stream {
    public static void main(String[] args) throws TwitterException {

    ConfigurationBuilder cb = new ConfigurationBuilder();
    cb.setDebugEnabled(true);
    cb.setOAuthConsumerKey("bbb");
    cb.setOAuthConsumerSecret("bbb");
    cb.setOAuthAccessToken("bbb");
    cb.setOAuthAccessTokenSecret("bbb");

    TwitterStream twitterStream = new TwitterStreamFactory(cb.build()).getInstance();
    StatusListener listener = new StatusListener() {

        public void onStatus(Status status) {
            System.out.println("@" + status.getUser().getScreenName() + " - " + status.getText());
        }

        public void onDeletionNotice(StatusDeletionNotice statusDeletionNotice) {
            System.out.println("Got a status deletion notice id:" + statusDeletionNotice.getStatusId());
        }

        public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
            System.out.println("Got track limitation notice:" + numberOfLimitedStatuses);
        }

        public void onScrubGeo(long userId, long upToStatusId) {
            System.out.println("Got scrub_geo event userId:" + userId + " upToStatusId:" + upToStatusId);
        }

        public void onException(Exception ex) {
            ex.printStackTrace();
        }
    };

    FilterQuery fq = new FilterQuery();
    String keywords[] = {"France", "Germany"};

    fq.track(keywords);

    twitterStream.addListener(listener);
    twitterStream.filter(fq);      
}
}