Rx Spotify Example Java version

package com.vertigo.crawler;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import com.google.common.collect.ImmutableList;
import com.wrapper.spotify.models.Album;
import com.wrapper.spotify.models.AlbumType;
import com.wrapper.spotify.models.Artist;
import com.wrapper.spotify.models.Copyright;
import com.wrapper.spotify.models.ExternalUrls;
import com.wrapper.spotify.models.Followers;
import com.wrapper.spotify.models.SimpleAlbum;
import com.wrapper.spotify.models.SimpleArtist;
import com.wrapper.spotify.models.SimpleTrack;
import com.wrapper.spotify.models.Track;
import rx.Observable;
import rx.Observer;

import com.vertigo.model.enums.EntityType;
import com.vertigo.model.enums.Provider;
import com.vertigo.model.json.sync.CrawlerTask;
import com.vertigo.test.util.RandomUtil;

public class Example {
    private final static Random       RANDOM = new Random(69);
    private final static RandomUtil   RU     = new RandomUtil(RANDOM);
    private final static List<String> GENRES = ImmutableList.of(
        "aggrotech", "ebm", "electro-industrial", "futurepop",
        "ambient psychill", "psychill", "ambient", "metal",
        "crossover thrash", "death metal", "groove metal",
        "nwobhm", "speed metal", "thrash metal", "digital hardcore",
        "industrial", "industrial metal", "industrial rock",
        "neue deutsche harte", "album rock", "alternative metal",
        "classic rock", "hard rock", "nu metal", "power metal",
        "rock", "wrestling");
    private final static List<String> MARKETS = ImmutableList.of(
        "AD", "AR", "AT", "AU", "BE", "BG", "BO", "BR", "CA", "CH",
        "CL", "CO", "CR", "CY", "CZ", "DE", "DK", "DO", "EC", "EE",
        "ES", "FI", "FR", "GB", "GR", "GT", "HK", "HN", "HU", "ID",
        "IE", "IS", "IT", "JP", "LI", "LT", "LU", "LV", "MC", "MT",
        "MX", "MY", "NI", "NL", "NO", "NZ", "PA", "PE", "PH", "PL",
        "PT", "PY", "SE", "SG", "SK", "SV", "TR", "TW", "US", "UY"
    );
    private final static Artist VA = get(() -> {
        Artist ar = new Artist();
        ar.setId("0LyfQWJT6nXafLPZqxe9Of");
        ar.setName("Various Artists");
        ar.setFollowers(get(() -> {
            Followers f = new Followers();
            f.setHref(null);
            f.setTotal(215138);
            return f;
        }));
        ar.setExternalUrls(get(() -> {
            ExternalUrls e = new ExternalUrls();
            e.getExternalUrls().put("spotify",
                "https://open.spotify.com/artist/" + ar.getId());
            return e;
        }));
        ar.setGenres(ImmutableList.of());
        ar.setHref("https://api.spotify.com/v1/artists/" + ar.getId());
        ar.setImages(ImmutableList.of());
        ar.setPopularity(0);
        ar.setUri("spotify:artist:" + ar.getId());
        return ar;
    });

    public static <T> T get(Supplier<T> supplier) {
        return supplier.get();
    }

    public static SimpleAlbum simplify(Album a) {
        SimpleAlbum sa = new SimpleAlbum();
        sa.setAlbumType(a.getAlbumType());
        sa.setExternalUrls(a.getExternalUrls());
        sa.setHref(a.getHref());
        sa.setId(a.getId());
        sa.setImages(a.getImages());
        sa.setName(a.getName());
        sa.setType(a.getType());
        sa.setUri(a.getUri());
        sa.setAvailableMarkets(ImmutableList.copyOf(a.getAvailableMarkets()));
        return sa;
    }

    public static SimpleTrack simplify(Track t) {
        SimpleTrack sa = new SimpleTrack();
        sa.setDiscNumber(t.getDiscNumber());
        sa.setDuration(t.getDuration());
        sa.setExplicit(t.isExplicit());
        sa.setExternalUrls(t.getExternalUrls());
        sa.setHref(t.getHref());
        sa.setId(t.getId());
        sa.setName(t.getName());
        sa.setPreviewUrl(t.getPreviewUrl());
        sa.setTrackNumber(t.getTrackNumber());
        sa.setType(t.getType());
        sa.setUri(t.getUri());
        sa.setAvailableMarkets(ImmutableList.copyOf(t.getAvailableMarkets()));
        return sa;
    }

    public static SimpleArtist simplify(Artist a) {
        SimpleArtist sa = new SimpleArtist();
        sa.setExternalUrls(a.getExternalUrls());
        sa.setHref(a.getHref());
        sa.setId(a.getId());
        sa.setName(a.getName());
        sa.setType(a.getType());
        sa.setUri(a.getUri());
        return sa;
    }

    public static Artist generateArtist(RandomUtil ru) {
        Artist a = new Artist();
        a.setId(ru.string(8));
        a.setName("artist_" + a.getId());
        a.setPopularity(ru.nextInt(100));
        a.setExternalUrls(get(() -> {
            ExternalUrls eu = new ExternalUrls();
            eu.getExternalUrls().put("spotify",
                "https://open.spotify.com/artist/" + a.getId());
            return eu;
        }));
        a.setHref("https://api.spotify.com/v1/artists/" + a.getId());
        a.setUri("spotify:artist:" + a.getId());
        a.setGenres(listFrom(ru, GENRES, 3));
        a.setFollowers(get(() -> {
            Followers f = new Followers();
            f.setTotal(ru.nextInt(1000));
            return f;
        }));
        return a;
    }

    public static Album generateAlbum(RandomUtil ru,
                                      AlbumType albumType,
                                      List<Artist> creators) {
        Album al = new Album();
        al.setId(ru.string(9));
        al.setName("album_" + al.getId());
        al.setAlbumType(albumType);
        al.setHref("https://api.spotify.com/v1/albums/" + al.getId());
        al.setUri("spotify:album:" + al.getId());
        al.setCopyrights(get(() -> {
            Copyright c1 = new Copyright();
            c1.setText("2016 Alpha Matrix");
            c1.setText("C");
            Copyright c2 = new Copyright();
            c1.setText("2016 Alpha Matrix");
            c1.setText("P");
            return ImmutableList.of(c1, c2);
        }));
        al.setArtists(get(() -> creators.stream()
            .map(Example::simplify)
            .collect(Collectors.toList())));
        return al;
    }

    public static Track generateTrack(RandomUtil ru,
                                      Album album,
                                      int trackNumber) {
        Track t = new Track();
        t.setId(ru.string(10));
        t.setName("track_" + t.getId());
        t.setHref("https://api.spotify.com/v1/tracks/" + t.getId());
        t.setUri("spotify:track:" + t.getId());
        t.setArtists(album.getArtists());
        t.setAlbum(simplify(album));
        t.setTrackNumber(trackNumber);
        // from 1 to 1000 seconds
        t.setDuration((ru.nextInt(1000) + 1) * 1000);
        t.setPopularity(ru.nextInt(10));
        t.setAvailableMarkets(album.getAvailableMarkets());
        t.setDiscNumber(0);
        t.setPreviewUrl("https://spotify.com/preview/" + t.getId());
        return t;
    }


    /**
     * Select random elements from base list.
     * Must be count <= base.size()
     */
    public static <T> List<T> listFrom(RandomUtil ru, List<T> base, int count) {
        int n = base.size();
        List<T> ts = new ArrayList<>(base);
        // remove random elements from there
        for (int i = 0; i < n - count; i++) {
            ts.remove(ru.nextInt(n - i));
        }
        return ts;
    }


    public static class RxWrapper {
        private final SpotifyWS sws;
        public RxWrapper(SpotifyWS sws) {this.sws = sws;}
        public Observable<Artist> getArtists(List<String> ids) {
            return Observable.from(sws.getArtists(ids));
        }
        public Observable<Album> getAlbums(List<String> ids) {
            return Observable.from(sws.getAlbums(ids));
        }
        public Observable<Track> getTracks(List<String> ids) {
            return Observable.from(sws.getTracks(ids));
        }
        public Observable<List<Track>> getTracks0(List<String> ids) {
            return Observable.just(sws.getTracks(ids));
        }
    }

    private static CrawlerTask randomTask(RandomUtil ru) {
        // set only entityType and externalId,
        // the rest of them should be populated by
        CrawlerTask ct = new CrawlerTask();
        ct.setEntityType(EntityType.fromInt(ru.nextInt(11, 13)));
        ct.setExternalId(ru.string(10));
        ct.setProvider(Provider.Spotify);
        return ct;
    }

    private static List<CrawlerTask> randomTasks(RandomUtil ru, int n) {
        return IntStream.range(0, n)
            .mapToObj(_i -> randomTask(ru))
            .collect(Collectors.toList());
    }

    private static List<String> ids(List<CrawlerTask> cts) {
        return cts.stream()
            .map(CrawlerTask::getExternalId)
            .collect(Collectors.toList());
    }

    interface SpotifyWS {
        List<Artist> getArtists(List<String> ids);
        List<Album> getAlbums(List<String> ids);
        List<Track> getTracks(List<String> ids);
    }

    /**
     *
     */
    public static class SpotifyWSImpl implements SpotifyWS {
        private final RandomUtil ru;
        private final Map<String, Track> trackMap = new HashMap<>();
        private final Map<String, Album> albumMap = new HashMap<>();
        private final Map<String, Artist> artistMap = new HashMap<>();

        public SpotifyWSImpl(RandomUtil ru) {
            this.ru = ru;
            initMediaCollection();
        }

        @Override
        public List<Artist> getArtists(List<String> ids) {
            int x = RU.nextInt(100);
            throw new RuntimeException("timeout:" + Integer.toString(x));
        }

        @Override
        public List<Album> getAlbums(List<String> ids) {
            int x = RU.nextInt(100);
            throw new RuntimeException("timeout:" + Integer.toString(x));
        }

        @Override
        public List<Track> getTracks(List<String> ids) {
            int x = RU.nextInt(100);
            throw new RuntimeException("timeout:" + Integer.toString(x));
        }

        private void initMediaCollection() {
            // create all
            List<Artist> artists = IntStream
                .range(0, 100)
                .mapToObj(_i -> generateArtist(ru))
                .collect(Collectors.toList());
            // let each artist to have 5 albums
            // 3 exclusive, 1 with other,
            // 1 with other two
            // and many album-compilations where
            // song is made by different artist
            // e.g. album 3m6QKbHMF5GVBagWlCF1U0
            Stream<Album> albums1 = IntStream
                .range(0, artists.size())
                .mapToObj(i -> {
                    Artist       ar  = artists.get(i);
                    List<Artist> ars = ImmutableList.of(ar);
                    artistMap.put(ar.getId(), ar);
                    Album al1 = generateAlbum(ru, AlbumType.SINGLE, ars);
                    Album al2 = generateAlbum(ru, AlbumType.SINGLE, ars);
                    Album al3 = generateAlbum(ru, AlbumType.SINGLE, ars);
                    return ImmutableList.of(al1, al2, al3).stream();
                })
                .flatMap(Function.identity());
            // create albums of pairs
            Collections.shuffle(artists);
            Stream<Album> albums2 = IntStream
                .range(1, artists.size())
                .mapToObj(i -> {
                    List<Artist> ars = ImmutableList.of(
                        artists.get(i - 1),
                        artists.get(i)
                    );
                    return generateAlbum(ru, AlbumType.ALBUM, ars);
                });
            // create albums of triples
            Collections.shuffle(artists);
            Stream<Album> albums3 = IntStream
                .range(2, artists.size())
                .mapToObj(i -> {
                    List<Artist> ars = ImmutableList.of(
                        artists.get(i - 2),
                        artists.get(i - 1),
                        artists.get(i)
                    );
                    return generateAlbum(ru, AlbumType.ALBUM, ars);
                });
            // create compilations,
            // albums where the artist is "Various Artist"
            Stream<Album> albums4 = IntStream
                .range(0, 200)
                .mapToObj(i -> {
                    List<Artist> ars = ImmutableList.of(VA);
                    return generateAlbum(ru, AlbumType.COMPILATION, ars);
                });
            // glue all the stuff together
            List<Album> albums = Stream
                .concat(albums1, Stream
                    .concat(albums2, Stream
                        .concat(albums3,
                            albums4)))
                    .collect(Collectors.toList());
            // now create tracks, let each album has 10 tracks
            Stream<Track> tracks = albums.stream()
                .flatMap(al ->
                    IntStream.range(0, 10)
                        .mapToObj(i ->
                            generateTrack(ru, al, i)));

            // now put the stuff into the maps
            artists.forEach(ar -> artistMap.put(ar.getId(), ar));
            albums.forEach(al -> albumMap.put(al.getId(), al));
            List<Track> tracksList = tracks.collect(Collectors.toList());
            tracksList.forEach(t -> trackMap.put(t.getId(), t));
        }

        private void sleep(int x) {
            try {
                Thread.sleep(10 * x);
            } catch (InterruptedException e) {
                System.out.println("e = " + e);
            }
        }
    }

    public static void main(String[] args) throws Exception {

        final SpotifyWSImpl sws = new SpotifyWSImpl(RU);
        final RxWrapper rxw = new RxWrapper(sws);

        List<CrawlerTask> tasks = randomTasks(RU, 1000);
        Observable.from(tasks)
            .filter(ct -> ct.getEntityType() == EntityType.Song)
            .buffer(2)
            .flatMap(cts-> rxw.getTracks(ids(cts)))
            .flatMap(t -> {
                // we got track, now we want the artists and album from it
                List<SimpleArtist> artists = t.getArtists();
                SimpleAlbum album = t.getAlbum();
                return Observable.just(t);
            })
            .subscribe(t -> new Observer<Track>() {
                @Override
                public void onCompleted() {
                    System.out.println("Completed t = " + t.getName());
                }
                @Override
                public void onError(Throwable e) {
                    System.out.println("Error t = " + t.getName() + " e = " + e);
                }
                @Override
                public void onNext(Track track) {
                    System.out.println("Next track = " + track);
                }
            });
    }
}