mirror of https://github.com/reiseburo/offtopic
parent
9e23d4f9b6
commit
4dc819f884
|
@ -44,6 +44,9 @@ dependencies {
|
||||||
compile 'org.apache.curator:curator-framework:[2.7.1,2.8)'
|
compile 'org.apache.curator:curator-framework:[2.7.1,2.8)'
|
||||||
compile 'org.apache.commons:commons-pool2:[2.2,3.0)'
|
compile 'org.apache.commons:commons-pool2:[2.2,3.0)'
|
||||||
|
|
||||||
|
/* For logging at runtime */
|
||||||
|
compile 'ch.qos.logback:logback-parent:[1.1.3,2.0)'
|
||||||
|
|
||||||
compile 'org.apache.kafka:kafka_2.11:0.8.2.1'
|
compile 'org.apache.kafka:kafka_2.11:0.8.2.1'
|
||||||
// Forcing us up to ZK 3.5 to prevent wacky classpath errors when mixing
|
// Forcing us up to ZK 3.5 to prevent wacky classpath errors when mixing
|
||||||
// and matching dependencies
|
// and matching dependencies
|
||||||
|
|
|
@ -1,9 +1,12 @@
|
||||||
package offtopic
|
package offtopic
|
||||||
|
|
||||||
|
import groovy.util.logging.Slf4j
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* class for wrapping our system configuration
|
* class for wrapping our system configuration
|
||||||
*/
|
*/
|
||||||
@Singleton
|
@Singleton
|
||||||
|
@Slf4j
|
||||||
class Configuration extends Properties {
|
class Configuration extends Properties {
|
||||||
/**
|
/**
|
||||||
* Load defaults, starting in the current working directory, searching for
|
* Load defaults, starting in the current working directory, searching for
|
||||||
|
@ -12,7 +15,11 @@ class Configuration extends Properties {
|
||||||
public boolean loadDefaults() {
|
public boolean loadDefaults() {
|
||||||
File cwdConfiguration = new File('offtopic.properties')
|
File cwdConfiguration = new File('offtopic.properties')
|
||||||
if (cwdConfiguration.exists()) {
|
if (cwdConfiguration.exists()) {
|
||||||
|
log.info('Loading `offtopic.properties` for configuration')
|
||||||
this.load(new FileInputStream(cwdConfiguration))
|
this.load(new FileInputStream(cwdConfiguration))
|
||||||
}
|
}
|
||||||
|
else {
|
||||||
|
log.warn('Could not load configuration file {}', cwdConfiguration.absolutePath)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,36 +1,40 @@
|
||||||
package offtopic
|
package offtopic
|
||||||
|
|
||||||
import groovy.json.JsonSlurper
|
import groovy.json.JsonSlurper
|
||||||
|
import groovy.util.logging.Slf4j
|
||||||
import offtopic.curator.CuratorPool
|
import offtopic.curator.CuratorPool
|
||||||
|
import org.apache.curator.framework.CuratorFramework
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* KafkaService is a helper class to help expose data about Kafka to the
|
* KafkaService is a helper class to help expose data about Kafka to the
|
||||||
* Ratpack app
|
* Ratpack app
|
||||||
*/
|
*/
|
||||||
|
@Slf4j
|
||||||
class KafkaService {
|
class KafkaService {
|
||||||
final static String BROKERS_PATH = "/brokers/ids"
|
final static String BROKERS_PATH = "/brokers/ids"
|
||||||
final static String TOPICS_PATH = "/brokers/topics"
|
final static String TOPICS_PATH = "/brokers/topics"
|
||||||
|
|
||||||
public static ArrayList fetchBrokers() {
|
public static List<String> fetchBrokers() {
|
||||||
def brokers = new ArrayList()
|
JsonSlurper parser = new JsonSlurper()
|
||||||
def parser = new JsonSlurper()
|
List<String> brokers = []
|
||||||
|
|
||||||
CuratorPool.withCurator { c ->
|
CuratorPool.withCurator { CuratorFramework c ->
|
||||||
c.getChildren().forPath(BROKERS_PATH).each { String id ->
|
c.children.forPath(BROKERS_PATH).each { String id ->
|
||||||
// Pulling this into a String buffer since parse(byte[]) is
|
// Pulling this into a String buffer since parse(byte[]) is
|
||||||
// throwing a stackoverflow error
|
// throwing a stackoverflow error
|
||||||
String buffer = new String(c.getData().forPath("${BROKERS_PATH}/${id}"))
|
String buffer = new String(c.data.forPath("${BROKERS_PATH}/${id}"))
|
||||||
brokers.add(parser.parseText(buffer))
|
brokers.add(parser.parseText(buffer))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
log.info('Fetched brokers from Zookeeper: {}', brokers)
|
||||||
return brokers
|
return brokers
|
||||||
}
|
}
|
||||||
|
|
||||||
public static ArrayList fetchTopics() {
|
public static List<String> fetchTopics() {
|
||||||
ArrayList brokers = null
|
List brokers = []
|
||||||
CuratorPool.withCurator { c ->
|
CuratorPool.withCurator { c ->
|
||||||
brokers = c.getChildren().forPath(TOPICS_PATH)
|
brokers = c.children.forPath(TOPICS_PATH)
|
||||||
}
|
}
|
||||||
return brokers
|
return brokers
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,10 +1,11 @@
|
||||||
package offtopic.curator
|
package offtopic.curator
|
||||||
|
|
||||||
|
import org.apache.curator.framework.CuratorFramework
|
||||||
import org.apache.curator.retry.ExponentialBackoffRetry
|
import org.apache.curator.retry.ExponentialBackoffRetry
|
||||||
import org.apache.curator.framework.CuratorFrameworkFactory
|
import org.apache.curator.framework.CuratorFrameworkFactory
|
||||||
|
|
||||||
class CuratorClient {
|
class CuratorClient {
|
||||||
def client = null
|
CuratorFramework client
|
||||||
|
|
||||||
public CuratorClient(String zookeepers) {
|
public CuratorClient(String zookeepers) {
|
||||||
if (zookeepers?.length() <= 0) {
|
if (zookeepers?.length() <= 0) {
|
||||||
|
|
|
@ -21,7 +21,7 @@ class CuratorPool extends GenericObjectPool<CuratorClient>{
|
||||||
}
|
}
|
||||||
|
|
||||||
public static void withCurator(Closure closure) {
|
public static void withCurator(Closure closure) {
|
||||||
def curator = null
|
CuratorClient curator = null
|
||||||
try {
|
try {
|
||||||
curator = CuratorPool.instance.borrowObject()
|
curator = CuratorPool.instance.borrowObject()
|
||||||
closure.call(curator.client)
|
closure.call(curator.client)
|
||||||
|
|
|
@ -1,13 +1,16 @@
|
||||||
|
import org.slf4j.Logger
|
||||||
|
import org.slf4j.LoggerFactory
|
||||||
import ratpack.handlebars.HandlebarsModule
|
import ratpack.handlebars.HandlebarsModule
|
||||||
import ratpack.jackson.JacksonModule
|
import ratpack.jackson.JacksonModule
|
||||||
|
|
||||||
|
import java.util.logging.Level
|
||||||
|
|
||||||
import static ratpack.handlebars.Template.handlebarsTemplate
|
import static ratpack.handlebars.Template.handlebarsTemplate
|
||||||
import static ratpack.jackson.Jackson.json
|
import static ratpack.jackson.Jackson.json
|
||||||
import static ratpack.groovy.Groovy.*
|
import static ratpack.groovy.Groovy.*
|
||||||
import static ratpack.websocket.WebSockets.websocket
|
import static ratpack.websocket.WebSockets.websocket
|
||||||
|
|
||||||
import offtopic.KafkaService
|
import offtopic.KafkaService
|
||||||
import offtopic.KafkaSubscriber
|
|
||||||
import offtopic.Configuration
|
import offtopic.Configuration
|
||||||
import offtopic.OfftopicClient
|
import offtopic.OfftopicClient
|
||||||
|
|
||||||
|
@ -26,7 +29,6 @@ ratpack {
|
||||||
|
|
||||||
get('topics') {
|
get('topics') {
|
||||||
topics = KafkaService.fetchTopics()
|
topics = KafkaService.fetchTopics()
|
||||||
println topics
|
|
||||||
|
|
||||||
if (request.headers.get('Content-Type') == 'application/json') {
|
if (request.headers.get('Content-Type') == 'application/json') {
|
||||||
render(json(topics))
|
render(json(topics))
|
||||||
|
@ -45,19 +47,20 @@ ratpack {
|
||||||
}
|
}
|
||||||
|
|
||||||
get('topics/:name/websocket') { ctx ->
|
get('topics/:name/websocket') { ctx ->
|
||||||
def client = new OfftopicClient(Configuration.instance)
|
OfftopicClient client = new OfftopicClient(Configuration.instance)
|
||||||
def grepper = null
|
String grepper = null
|
||||||
|
|
||||||
websocket(ctx) { ws ->
|
websocket(ctx) { ws ->
|
||||||
println "Connected ${ws}"
|
Logger log = LoggerFactory.getLogger('WebSocket')
|
||||||
|
log.info('Connected client {}', ws)
|
||||||
client.onMessageCallback = { m ->
|
client.onMessageCallback = { m ->
|
||||||
println "called back with ${m} (grep: ${grepper})"
|
log.debug('Callback with {] (grep: {})', m, grepper)
|
||||||
if ((grepper == null) || (m.raw =~ grepper)) {
|
if ((grepper == null) || (m.raw =~ grepper)) {
|
||||||
ws.send(groovy.json.JsonOutput.toJson(m))
|
ws.send(groovy.json.JsonOutput.toJson(m))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
client.createSubscribersFor(pathTokens.name)
|
client.createSubscribersFor(pathTokens.name)
|
||||||
print "subscribers created for ${pathTokens.name}"
|
log.info('Subscribers created for {}', pathTokens.name)
|
||||||
client.startSubscribers()
|
client.startSubscribers()
|
||||||
} connect { sock ->
|
} connect { sock ->
|
||||||
sock.onClose {
|
sock.onClose {
|
||||||
|
@ -80,18 +83,6 @@ ratpack {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/* set up a demo/dummy websocket listener */
|
|
||||||
get("ws") { context ->
|
|
||||||
websocket(context) { ws ->
|
|
||||||
} connect {
|
|
||||||
it.onClose {
|
|
||||||
println "closing"
|
|
||||||
} onMessage {
|
|
||||||
"client sent me ${it}"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
fileSystem "public", { f -> f.files() }
|
fileSystem "public", { f -> f.files() }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue