Instagram real-time Tags and Geographies subscription via Groovy and Ratpack https://instagram.com/developer/realtime/

instagram.groovy

@Grapes([
    @Grab('org.codehaus.gpars:gpars:1.2.1'),
    @Grab('com.github.groovy-wslite:groovy-wslite:1.1.0'),
    @Grab('io.ratpack:ratpack-groovy:0.9.15'),
    //@Grab('io.ratpack:ratpack-jackson:0.9.15'),
    @Grab('org.slf4j:slf4j-simple:1.7.12')])

import static groovyx.gpars.actor.Actors.actor
import static ratpack.groovy.Groovy.ratpack
import ratpack.http.client.HttpClient
import wslite.rest.*

String client = '' // obtain your own at https://instagram.com/developer
String secret = ''
String me = 'http://myapp.domain.com'
String myUpdates = "$me/updates"

String auth = "client_secret=${secret}&client_id=${client}"
String api = 'https://api.instagram.com/v1'
String apiSubs = "${api}/subscriptions"
URI instaSubs = new URI(apiSubs)
URI instaSubsAuth = new URI("${apiSubs}?${auth}")
URI instaSubsDel = new URI("${apiSubs}?${auth}&object=all")
Map pluralize = [ tag: 'tags', geography: 'geographies' ]
int count = 10 // count of tagged media to return
def instaRecent = { kind, oid -> "/${pluralize[kind]}/${oid}/media/recent?client_id=${client}&count=${count}" }
int radius = 5000 // meters, max allowed
String cities = 'cities15000.txt' // http://download.geonames.org/export/dump/
String tags = 'love instagood me tbt cute follow followme photooftheday happy tagforlikes beautiful girl like selfie picoftheday summer fun smile friends like4like instadaily fashion igers instalike food'
int limit = 15 // {"meta":{"error_type":"APINotAllowedError","code":400,"error_message":"You may only create 30 subscriptions"}}
int update = 10l // sec

String myToken = UUID.randomUUID().toString()
def js = new groovy.json.JsonSlurper()
def H = HttpClient.class

String _file(String file) { new String(new File(file).readBytes(), 'UTF-8') }

def objects = actor {
    def subscriptions = [:]
    def images = [] as Set
    loop {
        react { msg ->
            switch (msg.op) {
                case 'subscription':
                    println msg
                    subscriptions[msg.oid] = msg.title
                    break

                case 'media':
                    if (images.contains(msg.id)) {
                      //println "already saw ${subscriptions[msg.oid]} -> ${msg.id} ${msg.url}"
                    } else {
                        printf("%20s -> %-120s %s\n", subscriptions[msg.oid], msg.url, msg.location ?: '')
                        images.add(msg.id)
                    }
                    break
            }
        }
    }
}

def updater = actor {
    def pending = [] as Set
    def instagram = new RESTClient(api)
    loop {
        react { msg ->
          //println msg
            switch (msg.op) {
                case 'tick':
                    try {
                        pending.each { obj ->
                          //println obj
                            instagram.get(path: instaRecent(obj.kind, obj.oid)).json.data.each { media ->
                                objects << [ op: 'media', oid: obj.oid, id: media.id, url: media.images.thumbnail.url, location: media.location ]
                            }
                        }
                    } catch (Exception e) { println e.message }
                    pending.clear()
                    break

                case 'update':
                    pending.add(msg)
                    break
            }
        }
    }
}

groovyx.gpars.scheduler.Timer.timer.scheduleAtFixedRate({ updater << [ op: 'tick' ] }, update, update, java.util.concurrent.TimeUnit.SECONDS)

ratpack {
    handlers {
        get('subscribe') {
            def geographies = _file(cities).split('\n').collect { row ->
                    c = row.split('\t')
                    [ city: c[1], lat: c[4], lng: c[5], country: c[8], population: c[14] ]
                } .sort { it.population as int } .reverse().take(limit).collect { p ->
                    [ title: "${p.country} ${p.city}", params: "&object=geography&lat=${p.lat}&lng=${p.lng}&radius=${radius}" ]
                }
            def tags_ = tags.split(' ').take(limit).collect { tag ->
                    [ title: "#$tag", params: "&object=tag&object_id=$tag" ]
                }
            println 'subscribing to:\n' + (geographies + tags_).join('\n')
            def http = get(H)
            def futures = (geographies + tags_).collect { sub ->
                def resp = http.post(instaSubs) { req ->
                    req.body { body ->
                        body.type('application/x-www-form-urlencoded')
                            .text("${auth}&aspect=media&callback_url=${me}/updates&verify_token=${myToken}${sub.params}")
                    }
                }
                resp.map { r ->
                    def json = js.parseText(r.body.text)
                    if (json?.meta?.code == 200 && json?.data?.object_id)
                        objects << [ op: 'subscription', oid: json.data.object_id, title: sub.title ]
                    "${sub.title}: ${r.body.text}"
                } .mapError { ex -> "${sub.title}: ${ex.message}" }
            }
            render(futures.inject(blocking { 'subscribe:' }) { accf, subf -> accf.flatMap { acc -> subf.map { sub -> acc + "\n" + sub } } })
        }

        prefix('updates') {
            handler {
                byMethod {
                    post {
                        response.send 'great!'
                      //println '===== updates: ' + request.body.text
                        js.parseText(request.body.text).each { obj ->
                            updater << [ op: 'update', kind: obj.object, oid: obj.object_id ]
                        }
                    }

                    get {
                        def p = request.queryParams
                        if (p['hub.mode'] == 'subscribe' && p['hub.verify_token'] == myToken) {
                            println 'subscription confirmed'
                            response.send p['hub.challenge']
                        } else {
                            println "GET /updates: $p"
                            response.send 'what?'
                        }
                    }
                }
            }
        }

        get('list') {
            render(get(H).get(instaSubsAuth).map { it.body.text })
        }

        get('unsubscribe') {
            render(get(H).request(instaSubsDel) { it.method('DELETE') } .map { it.body.text })
        }

        get('oauth') {
        }

        post('ui') {
            println 'ui: ' + request.body.text
            response.send()
        }
    }
}