@Override
    public List<Object> filter(Status status) {
        URLEntity urls[] = status.getURLEntities();

        if (urls == null) {
            return null;
        }

        URL finalUrl = null;
        List<Object> marketUrls = new LinkedList<Object>();
        for (URLEntity url : urls) {
            try {
                //finalUrl = getFinalUrl(url.getURL());
                finalUrl = new URL(url.getExpandedURL());

                if (finalUrl == null) {
                    continue;
                }

                String extra = null;
                if (!url.getURL().toString().equals(finalUrl.toString())) {
                    extra = ", \"original\": \"" + url.getURL().toString() + "\"";
                }

                JSONObject msg = new JSONObject();
                msg.put("link", finalUrl.toString());
                msg.put("host", finalUrl.getHost());
                if (extra != null) {
                    msg.put("original", url.getURL().toString());
                }

                publish(msg.toJSONString());

                if ("play.google.com".equals(finalUrl.getHost()) && finalUrl.getPath().contains("details")) {
                    marketUrls.add(msg.toJSONString());
                }
            } catch (MalformedURLException ex) {
                Logger.getLogger(RedisLinksPublisherBolt.class.getName()).log(Level.SEVERE, null, ex);
            }
        }


        return marketUrls.size() == 0 ? null : marketUrls;

    }