The RSS reader tutorial

This tu­to­r­ial is ded­i­cated for users who’d like to know how to use the Eclipse Vert.x Cas­san­dra client in prac­tice.

Before you start this tutorial

Be­fore start­ing, you should

• be fa­mil­iar with Eclipse Vert.x. If you are not, here is the guide for Java de­vel­op­ers.

• have some basic un­der­stand­ing of data­bases. Ex­tended knowl­edge of Cas­san­dra is not re­quired.

You also may find it use­ful to read the RSS 2.0 spec­i­fi­ca­tion, be­cause the re­sulted app is, ba­si­cally, a stor­age of RSS 2.0 feeds.

To give you an idea of what the App is about, here is how it looks like from the fronted side:

In the image, we see that browser space is split into 2 parts:

1. Saved feed names
2. List of ar­ti­cles for the se­lected feed

Here you also can enter a link to a new feed, so the App will fetch and parse the feed. After that, it will ap­pear in the left col­umn along with other saved feeds.

Requirements

For com­plet­ing this tu­to­r­ial you need:

• Java 8 or higher
• Git
• 1 hour of your time
• You fa­vorite code ed­i­tor

For run­ning the ex­am­ple you should en­sure that Cas­san­dra ser­vice is run­ning lo­cally on port 9042. As an op­tion, you can run Cas­san­dra with ccm(Cas­san­dra Clus­ter Man­ager). Fol­low this in­struc­tions for in­stalling ccm. After in­stalling you will be able to run a sin­gle node clus­ter:

$ ccm create rss_reader -v 3.11.2 -n 1 -s
$ ccm start

Be­fore com­plet­ing this step make sure that you have suc­cess­fully cloned the RSS reader repos­i­tory and checked out the step_1 branch:

$ git clone https://github.com/Sammers21/rss-reader
$ cd rss-reader
$ git checkout step_1

Now you can try to tun this ex­am­ple and see if it works:

$ ./gradlew vertxRun

Schema

If you are fa­mil­iar with Apache Cas­san­dra, you should know that the way your data is stored in Cas­san­dra is de­pen­dent on queries you are run­ning. It means that you need first to fig­ure out what kind of queries you will be run­ning, and then you can pro­duce a stor­age scheme.

In our case, we’d like our ap­pli­ca­tion to have 3 end­points:

1. POST /user/{user_id}/rss_link - for adding links to a user’s feed
2. GET /user/{user_id}/rss_channels - for re­triev­ing in­for­ma­tion about RSS chan­nels a user sub­scribed on
3. GET /articles/by_rss_link?link={rss_link} - for re­triev­ing in­for­ma­tion about ar­ti­cles on a spe­cific RSS chan­nel

For im­ple­ment­ing this end­points, the schema should look as fol­lows:

CREATE TABLE rss_by_user (login text , rss_link text, PRIMARY KEY (login, rss_link));
CREATE TABLE articles_by_rss_link(rss_link text, pubDate timestamp, title text, article_link text, description text, PRIMARY KEY ( rss_link , pubDate , article_link));
CREATE TABLE channel_info_by_rss_link(rss_link text, last_fetch_time timestamp,title text, site_link text, description text, PRIMARY KEY(rss_link));

What to do in this step

In this step, we will im­ple­ment only the first end­point

Project overview

There are two no­table classes in the project: AppVerticle and FetchVerticle. The first one is a Ver­ti­cle re­spon­si­ble for HTTP re­quest han­dling and stor­age schema ini­tial­iza­tion. The sec­ond one is a Ver­ti­cle as well, but re­spon­si­ble for RSS feeds fetch­ing.

The idea is sim­ple. When the ap­pli­ca­tion is start­ing the AppVerticle is de­ployed, then it tries to ini­tial­ize stor­age schema, de­scribed in src/main/resources/schema.cql file by read­ing it and ex­e­cut­ing listed queries line by line. After the schema ini­tial­iza­tion the AppVerticle de­ploys FetchVerticle and starts a HTTP server.

Implementing the endpoint

Now, it is time to im­ple­ment the first end­point. Pay at­ten­tion to TODOs, they are for point­ing you out about where changes should be made.

Now, let’s have a look at the AppVerticle#postRssLink method. This method is called each time the first end­point is called, so we can fig­ure out what is the posted body and id of the user, who per­formed the re­quest, di­rectly there. There are 2 main things we want to do in this method:

1. No­ti­fy­ing via the Event Bus the FetchVerticle to fetch given by user link link to an RSS feed.
2. In­sert­ing an entry to the rss_by_user table.

This is how the AppVerticle#postRssLink method should be im­ple­mented:

private void postRssLink(RoutingContext ctx) {
    ctx.request().bodyHandler(body -> {
        JsonObject bodyAsJson = body.toJsonObject();
        String link = bodyAsJson.getString("link");
        String userId = ctx.request().getParam("user_id");
        if (link == null || userId == null) {
            responseWithInvalidRequest(ctx);
        } else {
            vertx.eventBus().send("fetch.rss.link", link);
            Future<ResultSet> future = Future.future();
            BoundStatement query = insertNewLinkForUser.bind(userId, link);
            client.execute(query, future);
            future.setHandler(result -> {
                if (result.succeeded()) {
                    ctx.response().end(new JsonObject().put("message", "The feed just added").toString());
                } else {
                    ctx.response().setStatusCode(400).end(result.cause().getMessage());
                }
            });
        }
    });
}

private void responseWithInvalidRequest(RoutingContext ctx) {
    ctx.response()
            .setStatusCode(400)
            .putHeader("content-type", "application/json; charset=utf-8")
            .end(invalidRequest().toString());
}

private JsonObject invalidRequest() {
    return new JsonObject().put("message", "Invalid request");
}

You may no­tice that insertNewLinkForUser is a PreparedStatement, and should be ini­tial­ized be­fore the AppVerticle start. Let’s do it in the AppVerticle#prepareNecessaryQueries method:

private Future<Void> prepareNecessaryQueries() {
    Future<PreparedStatement> insertNewLinkForUserPrepFuture = Future.future();
    client.prepare("INSERT INTO rss_by_user (login , rss_link ) VALUES ( ?, ?);", insertNewLinkForUserPrepFuture);

    return insertNewLinkForUserPrepFuture.compose(preparedStatement -> {
        insertNewLinkForUser = preparedStatement;
        return Future.succeededFuture();
    });
}

Also, we should not for­get to fetch a RSS by the link sent to FetchVerticle via the Event Bus. We can do it in the FetchVerticle#startFetchEventBusConsumer method:

vertx.eventBus().localConsumer("fetch.rss.link", message -> {
    String rssLink = (String) message.body();
    log.info("fetching " + rssLink);
    webClient.getAbs(rssLink).send(response -> {
        if (response.succeeded()) {
            String bodyAsString = response.result().bodyAsString("UTF-8");
            try {
                RssChannel rssChannel = new RssChannel(bodyAsString);

                BatchStatement batchStatement = new BatchStatement();
                BoundStatement channelInfoInsertQuery = insertChannelInfo.bind(
                        rssLink, new Date(System.currentTimeMillis()), rssChannel.description, rssChannel.link, rssChannel.title
                );
                batchStatement.add(channelInfoInsertQuery);

                for (Article article : rssChannel.articles) {
                    batchStatement.add(insertArticleInfo.bind(rssLink, article.pubDate, article.link, article.description, article.title));
                }
                Future<ResultSet> insertArticlesFuture = Future.future();
                cassandraClient.execute(batchStatement, insertArticlesFuture);

                insertArticlesFuture.compose(insertDone -> Future.succeededFuture());
            } catch (Exception e) {
                log.error("Unable to fetch: " + rssLink, e);
            }
        } else {
            log.error("Unable to fetch: " + rssLink);
        }
    });
});

And, fi­nally, this code would not work if insertChannelInfo and insertArticleInfo state­ments will not be ini­tial­ized at ver­ti­cle start. Let’s to this in the FetchVerticle#prepareNecessaryQueries method:

 private Future<Void> prepareNecessaryQueries() {
        Future<PreparedStatement> insertChannelInfoPrepFuture = Future.future();
        cassandraClient.prepare("INSERT INTO channel_info_by_rss_link ( rss_link , last_fetch_time, description , site_link , title ) VALUES (?, ?, ?, ?, ?);", insertChannelInfoPrepFuture);

        Future<PreparedStatement> insertArticleInfoPrepFuture = Future.future();
        cassandraClient.prepare("INSERT INTO articles_by_rss_link ( rss_link , pubdate , article_link , description , title ) VALUES ( ?, ?, ?, ?, ?);", insertArticleInfoPrepFuture);

        return CompositeFuture.all(
                insertChannelInfoPrepFuture.compose(preparedStatement -> {
                    insertChannelInfo = preparedStatement;
                    return Future.succeededFuture();
                }), insertArticleInfoPrepFuture.compose(preparedStatement -> {
                    insertArticleInfo = preparedStatement;
                    return Future.succeededFuture();
                })
        ).mapEmpty();
    }

Observing

After all these changes, you should en­sure that the first end­point is work­ing cor­rectly. You need to run the ap­pli­ca­tion, go to lo­cal­host:8080 in­sert a link to a rss feed there(BBC UK feed news for ex­am­ple) and then click the ENTER but­ton. Now you can con­nect to your local Cas­san­dra in­stance, for in­stance with cqlsh, and find out how RSS feed data had been saved in the rss_reader key­space:

cqlsh> SELECT * FROM rss_reader.rss_by_user limit 1  ;

 login | rss_link
-------+-----------------------------------------
 Pavel | http://feeds.bbci.co.uk/news/uk/rss.xml

(1 rows)
cqlsh> SELECT description FROM rss_reader.articles_by_rss_link  limit 1;

 description
-------------------------------------
 BBC coverage of latest developments

(1 rows)

Conclusion

In this ar­ti­cle we fig­ured out how to im­ple­ment the first end­point of RSS-​reader app. If you have any prob­lems with com­plet­ing this step you can check­out to step_2, where you can find all changes made for com­plet­ing this step:

$ git checkout step_2

Thanks for read­ing this. I hope you en­joyed read­ing this ar­ti­cle. See you soon on our Git­ter chan­nel!