Now we can do the actual implementation for the search. First we want to add a list of feeds to the Searcher class, and also a dispatcher and a document factory to retrieve the articles:

class Searcher() {

    val dispatcher = newFixedThreadPoolContext(3, "IO-Search")
    val factory = DocumentBuilderFactory.newInstance()

    val feeds = listOf(
            Feed("npr", "https://www.npr.org/rss/rss.php?id=1001"),
            Feed("cnn", "http://rss.cnn.com/rss/cnn_topstories.rss"),
            Feed("fox", "http://feeds.foxnews.com/foxnews/latest?format=xml")
    )
    ...
}

Now we can add a search() function that will retrieve a feed, filter the articles containing the given query on the title or the description, and send them through a SendChannel received as a parameter. The initial implementation will be quite similar to the one we have used in the producer before:

private suspend fun search(
        feed: Feed,
        channel: SendChannel<Article>,
        query: String) {
    val builder = factory.newDocumentBuilder()
    val xml = builder.parse(feed.url)
    val news = xml.getElementsByTagName("channel").item(0)

    (0 until news.childNodes.length)
            .map { news.childNodes.item(it) }
            .filter { Node.ELEMENT_NODE == it.nodeType }
            .map { it as Element }
            .filter { "item" == it.tagName }
            .forEach {
                // TODO: Parse and filter
            }

}

But instead of mapping the content, we want to send it through a channel after filtering:

val title = it.getElementsByTagName("title")
        .item(0)
        .textContent
var summary = it.getElementsByTagName("description")
        .item(0)
        .textContent

if (title.contains(query) || summary.contains(query)) {
    if (summary.contains("<div")) {
        summary = summary.substring(0, summary.indexOf("<div"))
    }

    val article = Article(feed.name, title, summary)
    channel.send(article)
}