http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ed671149/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java b/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java deleted file mode 100644 index 0fb9828..0000000 --- a/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java +++ /dev/null @@ -1,292 +0,0 @@ -package org.apache.blur.thrift; - -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -import static org.apache.blur.utils.BlurConstants.BLUR_CLUSTER; -import static org.apache.blur.utils.BlurConstants.BLUR_CLUSTER_NAME; -import static org.apache.blur.utils.BlurConstants.BLUR_CONTROLLER_BIND_PORT; -import static org.apache.blur.utils.BlurConstants.BLUR_GUI_CONTROLLER_PORT; -import static org.apache.blur.utils.BlurConstants.BLUR_GUI_SHARD_PORT; -import static org.apache.blur.utils.BlurConstants.BLUR_INDEXMANAGER_SEARCH_THREAD_COUNT; -import static org.apache.blur.utils.BlurConstants.BLUR_MAX_CLAUSE_COUNT; -import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_BIND_ADDRESS; -import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_BIND_PORT; -import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_BLOCKCACHE_DIRECT_MEMORY_ALLOCATION; -import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_BLOCKCACHE_SLAB_COUNT; -import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_FILTER_CACHE_CLASS; -import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_HOSTNAME; -import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_INDEX_WARMUP_CLASS; -import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_OPENER_THREAD_COUNT; -import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_SAFEMODEDELAY; -import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_SERVER_THRIFT_THREAD_COUNT; -import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_TIME_BETWEEN_COMMITS; -import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_TIME_BETWEEN_REFRESHS; -import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_CONNECTION; -import static org.apache.blur.utils.BlurConstants.BLUR_ZOOKEEPER_SYSTEM_TIME_TOLERANCE; -import static org.apache.blur.utils.BlurUtil.quietClose; - -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; -import java.util.Map; - -import org.apache.blur.BlurConfiguration; -import org.apache.blur.concurrent.SimpleUncaughtExceptionHandler; -import org.apache.blur.concurrent.ThreadWatcher; -import org.apache.blur.gui.HttpJettyServer; -import org.apache.blur.log.Log; -import org.apache.blur.log.LogFactory; -import org.apache.blur.manager.BlurFilterCache; -import org.apache.blur.manager.DefaultBlurFilterCache; -import org.apache.blur.manager.IndexManager; -import org.apache.blur.manager.clusterstatus.ZookeeperClusterStatus; -import org.apache.blur.manager.indexserver.BlurIndexWarmup; -import org.apache.blur.manager.indexserver.BlurServerShutDown; -import org.apache.blur.manager.indexserver.BlurServerShutDown.BlurShutdown; -import org.apache.blur.manager.indexserver.DefaultBlurIndexWarmup; -import org.apache.blur.manager.indexserver.DistributedIndexServer; -import org.apache.blur.manager.indexserver.DistributedLayoutManager; -import org.apache.blur.manager.writer.BlurIndexRefresher; -import org.apache.blur.metrics.BlurMetrics; -import org.apache.blur.store.blockcache.BlockCache; -import org.apache.blur.store.blockcache.BlockDirectory; -import org.apache.blur.store.blockcache.BlockDirectoryCache; -import org.apache.blur.store.blockcache.Cache; -import org.apache.blur.store.buffer.BufferStore; -import org.apache.blur.thrift.generated.Blur.Iface; -import org.apache.blur.thrift.generated.Blur.Processor; -import org.apache.blur.utils.BlurConstants; -import org.apache.blur.utils.BlurUtil; -import org.apache.blur.zookeeper.ZkUtils; -import org.apache.hadoop.conf.Configuration; -import org.apache.lucene.index.IndexDeletionPolicy; -import org.apache.lucene.index.KeepOnlyLastCommitDeletionPolicy; -import org.apache.zookeeper.KeeperException; -import org.apache.zookeeper.KeeperException.Code; -import org.apache.zookeeper.ZooKeeper; - -public class ThriftBlurShardServer extends ThriftServer { - - private static final Log LOG = LogFactory.getLog(ThriftBlurShardServer.class); - - public static void main(String[] args) throws Exception { - int serverIndex = getServerIndex(args); - LOG.info("Setting up Shard Server"); - - Thread.setDefaultUncaughtExceptionHandler(new SimpleUncaughtExceptionHandler()); - BlurConfiguration configuration = new BlurConfiguration(); - - ThriftServer server = createServer(serverIndex, configuration); - server.start(); - } - - public static ThriftServer createServer(int serverIndex, BlurConfiguration configuration) throws Exception { - // setup block cache - // 134,217,728 is the slab size, therefore there are 16,384 blocks - // in a slab when using a block size of 8,192 - int numberOfBlocksPerSlab = 16384; - int blockSize = BlockDirectory.BLOCK_SIZE; - int slabCount = configuration.getInt(BLUR_SHARD_BLOCKCACHE_SLAB_COUNT, 1); - Cache cache; - Configuration config = new Configuration(); - - String bindAddress = configuration.get(BLUR_SHARD_BIND_ADDRESS); - int bindPort = configuration.getInt(BLUR_SHARD_BIND_PORT, -1); - bindPort += serverIndex; - - BlurMetrics blurMetrics = new BlurMetrics(config); - - int baseGuiPort = Integer.parseInt(configuration.get(BLUR_GUI_SHARD_PORT)); - final HttpJettyServer httpServer; - if (baseGuiPort > 0) { - int webServerPort = baseGuiPort + serverIndex; - - // TODO: this got ugly, there has to be a better way to handle all these - // params - // without reversing the mvn dependancy and making blur-gui on top. - httpServer = new HttpJettyServer(bindPort, webServerPort, configuration.getInt(BLUR_CONTROLLER_BIND_PORT, -1), configuration.getInt(BLUR_SHARD_BIND_PORT, -1), - configuration.getInt(BLUR_GUI_CONTROLLER_PORT, -1), configuration.getInt(BLUR_GUI_SHARD_PORT, -1), "shard", blurMetrics); - } else { - httpServer = null; - } - - if (slabCount >= 1) { - BlockCache blockCache; - boolean directAllocation = configuration.getBoolean(BLUR_SHARD_BLOCKCACHE_DIRECT_MEMORY_ALLOCATION, true); - - int slabSize = numberOfBlocksPerSlab * blockSize; - LOG.info("Number of slabs of block cache [{0}] with direct memory allocation set to [{1}]", slabCount, directAllocation); - LOG.info("Block cache target memory usage, slab size of [{0}] will allocate [{1}] slabs and use ~[{2}] bytes", slabSize, slabCount, ((long) slabCount * (long) slabSize)); - - BufferStore.init(configuration, blurMetrics); - - try { - long totalMemory = (long) slabCount * (long) numberOfBlocksPerSlab * (long) blockSize; - blockCache = new BlockCache(directAllocation, totalMemory, slabSize); - } catch (OutOfMemoryError e) { - if ("Direct buffer memory".equals(e.getMessage())) { - System.err - .println("The max direct memory is too low. Either increase by setting (-XX:MaxDirectMemorySize=<size>g -XX:+UseLargePages) or disable direct allocation by (blur.shard.blockcache.direct.memory.allocation=false) in blur-site.properties"); - System.exit(1); - } - throw e; - } - cache = new BlockDirectoryCache(blockCache); - } else { - cache = BlockDirectory.NO_CACHE; - } - - LOG.info("Shard Server using index [{0}] bind address [{1}]", serverIndex, bindAddress + ":" + bindPort); - - String nodeNameHostName = getNodeName(configuration, BLUR_SHARD_HOSTNAME); - String nodeName = nodeNameHostName + ":" + bindPort; - String zkConnectionStr = isEmpty(configuration.get(BLUR_ZOOKEEPER_CONNECTION), BLUR_ZOOKEEPER_CONNECTION); - - final ZooKeeper zooKeeper = ZkUtils.newZooKeeper(zkConnectionStr); - try { - ZookeeperSystemTime.checkSystemTime(zooKeeper, configuration.getLong(BLUR_ZOOKEEPER_SYSTEM_TIME_TOLERANCE, 3000)); - } catch (KeeperException e) { - if (e.code() == Code.CONNECTIONLOSS) { - System.err.println("Cannot connect zookeeper to [" + zkConnectionStr + "]"); - System.exit(1); - } - } - - BlurUtil.setupZookeeper(zooKeeper, configuration.get(BLUR_CLUSTER_NAME)); - - final ZookeeperClusterStatus clusterStatus = new ZookeeperClusterStatus(BlurConstants.BLUR_CLUSTER, zooKeeper); - - final BlurIndexRefresher refresher = new BlurIndexRefresher(); - refresher.init(); - - BlurFilterCache filterCache = getFilterCache(configuration); - BlurIndexWarmup indexWarmup = getIndexWarmup(configuration); - IndexDeletionPolicy indexDeletionPolicy = new KeepOnlyLastCommitDeletionPolicy(); - - final DistributedIndexServer indexServer = new DistributedIndexServer(); - indexServer.setBlurMetrics(blurMetrics); - indexServer.setCache(cache); - indexServer.setClusterStatus(clusterStatus); - indexServer.setClusterName(configuration.get(BLUR_CLUSTER_NAME, BLUR_CLUSTER)); - indexServer.setConfiguration(config); - indexServer.setNodeName(nodeName); - indexServer.setRefresher(refresher); - indexServer.setShardOpenerThreadCount(configuration.getInt(BLUR_SHARD_OPENER_THREAD_COUNT, 16)); - indexServer.setZookeeper(zooKeeper); - indexServer.setFilterCache(filterCache); - indexServer.setSafeModeDelay(configuration.getLong(BLUR_SHARD_SAFEMODEDELAY, 60000)); - indexServer.setWarmup(indexWarmup); - indexServer.setIndexDeletionPolicy(indexDeletionPolicy); - indexServer.setTimeBetweenCommits(configuration.getLong(BLUR_SHARD_TIME_BETWEEN_COMMITS, 60000)); - indexServer.setTimeBetweenRefreshs(configuration.getLong(BLUR_SHARD_TIME_BETWEEN_REFRESHS, 500)); - indexServer.init(); - - final IndexManager indexManager = new IndexManager(); - indexManager.setIndexServer(indexServer); - indexManager.setMaxClauseCount(configuration.getInt(BLUR_MAX_CLAUSE_COUNT, 1024)); - indexManager.setThreadCount(configuration.getInt(BLUR_INDEXMANAGER_SEARCH_THREAD_COUNT, 32)); - indexManager.setBlurMetrics(blurMetrics); - indexManager.setFilterCache(filterCache); - indexManager.init(); - - TableLayout layout = new TableLayout() { - @Override - public String findServer(String table, int shard, TYPE type) { - DistributedLayoutManager manager = new DistributedLayoutManager(); - List<String> onlineServers = clusterStatus.getOnlineServers(true); - List<String> offlineServers = clusterStatus.getOfflineServers(true); - manager.setNodes(onlineServers); - manager.setNodesOffline(offlineServers); - manager.setShards(getShardList(clusterStatus.getTableDescriptor(true, table).getShardCount())); - manager.init(); - - Map<String, String> map = manager.getLayout(); - String server = map.get(BlurUtil.getShardName(BlurConstants.SHARD_PREFIX, shard)); - return server; - } - - private Collection<String> getShardList(int shardCount) { - List<String> list = new ArrayList<String>(); - for (int i = 0; i < shardCount; i++) { - list.add(BlurUtil.getShardName(BlurConstants.SHARD_PREFIX, i)); - } - return list; - } - }; - - final BlurShardServer shardServer = new BlurShardServer(); - shardServer.setIndexServer(indexServer); - shardServer.setIndexManager(indexManager); - shardServer.setZookeeper(zooKeeper); - shardServer.setClusterStatus(clusterStatus); - shardServer.setConfiguration(configuration); - shardServer.setLayout(layout); - shardServer.init(); - - Iface iface = BlurUtil.recordMethodCallsAndAverageTimes(blurMetrics, shardServer, Iface.class); - - int threadCount = configuration.getInt(BLUR_SHARD_SERVER_THRIFT_THREAD_COUNT, 32); - - Processor<Iface> processor = new Processor<Iface>(iface); - - final ThriftBlurShardServer server = new ThriftBlurShardServer(); - server.setNodeName(nodeName); - server.setBindAddress(bindAddress); - server.setBindPort(bindPort); - server.setThreadCount(threadCount); - server.setConfiguration(configuration); - server.setProcessor(processor); - - // This will shutdown the server when the correct path is set in zk - BlurShutdown shutdown = new BlurShutdown() { - @Override - public void shutdown() { - ThreadWatcher threadWatcher = ThreadWatcher.instance(); - quietClose(refresher, server, shardServer, indexManager, indexServer, threadWatcher, clusterStatus, zooKeeper, httpServer); - } - }; - server.setShutdown(shutdown); - new BlurServerShutDown().register(shutdown, zooKeeper); - return server; - } - - private static BlurFilterCache getFilterCache(BlurConfiguration configuration) { - String _blurFilterCacheClass = configuration.get(BLUR_SHARD_FILTER_CACHE_CLASS); - if (_blurFilterCacheClass != null) { - try { - Class<?> clazz = Class.forName(_blurFilterCacheClass); - return (BlurFilterCache) clazz.newInstance(); - } catch (Exception e) { - throw new RuntimeException(e); - } - } - return new DefaultBlurFilterCache(); - } - - private static BlurIndexWarmup getIndexWarmup(BlurConfiguration configuration) { - String _blurFilterCacheClass = configuration.get(BLUR_SHARD_INDEX_WARMUP_CLASS); - if (_blurFilterCacheClass != null) { - try { - Class<?> clazz = Class.forName(_blurFilterCacheClass); - return (BlurIndexWarmup) clazz.newInstance(); - } catch (Exception e) { - throw new RuntimeException(e); - } - } - return new DefaultBlurIndexWarmup(); - } -}
http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ed671149/src/blur-core/src/test/java/org/apache/blur/MiniCluster.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/test/java/org/apache/blur/MiniCluster.java b/src/blur-core/src/test/java/org/apache/blur/MiniCluster.java index 777d880..4f89e12 100644 --- a/src/blur-core/src/test/java/org/apache/blur/MiniCluster.java +++ b/src/blur-core/src/test/java/org/apache/blur/MiniCluster.java @@ -45,7 +45,7 @@ import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; import org.apache.blur.thrift.BlurClientManager; import org.apache.blur.thrift.Connection; -import org.apache.blur.thrift.ThriftBlurShardServer; +import org.apache.blur.thrift.ThriftBlurServer; import org.apache.blur.thrift.ThriftServer; import org.apache.blur.thrift.generated.Blur.Client; import org.apache.commons.lang.StringUtils; @@ -140,7 +140,7 @@ public abstract class MiniCluster { futures.add(executorService.submit(new Callable<ThriftServer>() { @Override public ThriftServer call() throws Exception { - return ThriftBlurShardServer.createServer(index, localConf); + return ThriftBlurServer.createServer(index, localConf); } })); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ed671149/src/blur-core/src/test/java/org/apache/blur/search/RandomSuperQueryTest.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/test/java/org/apache/blur/search/RandomSuperQueryTest.java b/src/blur-core/src/test/java/org/apache/blur/search/RandomSuperQueryTest.java index 4c9d7d1..7dc9c71 100644 --- a/src/blur-core/src/test/java/org/apache/blur/search/RandomSuperQueryTest.java +++ b/src/blur-core/src/test/java/org/apache/blur/search/RandomSuperQueryTest.java @@ -18,15 +18,26 @@ package org.apache.blur.search; */ import static junit.framework.Assert.assertTrue; +import static org.apache.blur.lucene.LuceneVersionConstant.LUCENE_VERSION; import java.io.IOException; +import java.util.ArrayList; import java.util.Collection; +import java.util.HashMap; import java.util.HashSet; +import java.util.List; +import java.util.Map; import java.util.Random; +import org.apache.lucene.analysis.standard.StandardAnalyzer; +import org.apache.lucene.document.Document; +import org.apache.lucene.document.Field.Store; +import org.apache.lucene.document.StringField; import org.apache.lucene.index.CorruptIndexException; import org.apache.lucene.index.DirectoryReader; import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.IndexWriterConfig; import org.apache.lucene.queryparser.classic.ParseException; import org.apache.lucene.search.Filter; import org.apache.lucene.search.IndexSearcher; @@ -39,7 +50,6 @@ import org.apache.lucene.store.LockObtainFailedException; import org.apache.lucene.store.RAMDirectory; import org.junit.Test; - public class RandomSuperQueryTest { private static final int MOD_COLS_USED_FOR_SKIPPING = 3; @@ -72,7 +82,9 @@ public class RandomSuperQueryTest { IndexSearcher searcher = new IndexSearcher(reader); long s = System.currentTimeMillis(); for (String str : sampler) { - Query query = null;// new SuperParser(LUCENE_VERSION, new BlurAnalyzer(), true, filter, ScoreType.AGGREGATE).parse(str); + Query query = null;// new SuperParser(LUCENE_VERSION, new BlurAnalyzer(), + // true, filter, ScoreType.AGGREGATE).parse(str); + TopDocs topDocs = searcher.search(query, 10); assertTrue("seed [" + seed + "] {" + query + "} {" + s + "}", topDocs.totalHits > 0); } @@ -82,18 +94,17 @@ public class RandomSuperQueryTest { private Directory createIndex(Random random, Collection<String> sampler) throws CorruptIndexException, LockObtainFailedException, IOException { Directory directory = new RAMDirectory(); -// String[] columnFamilies = genWords(random, MIN_NUM_COL_FAM, MAX_NUM_COL_FAM, "colfam"); -// Map<String, String[]> columns = new HashMap<String, String[]>(); -// for (int i = 0; i < columnFamilies.length; i++) { -// columns.put(columnFamilies[i], genWords(random, MIN_NUM_COLS, MAX_NUM_COLS, "col")); -// } -// IndexWriter writer = new IndexWriter(directory, new IndexWriterConfig(LUCENE_VERSION, new StandardAnalyzer(LUCENE_VERSION))); -// RowIndexWriter indexWriter = new RowIndexWriter(writer, new BlurAnalyzer(new StandardAnalyzer(LUCENE_VERSION))); -// int numberOfDocs = random.nextInt(MAX_NUM_OF_DOCS) + 1; -// for (int i = 0; i < numberOfDocs; i++) { -// indexWriter.replace(false, generatSuperDoc(random, columns, sampler)); -// } -// writer.close(); + String[] columnFamilies = genWords(random, MIN_NUM_COL_FAM, MAX_NUM_COL_FAM, "colfam"); + Map<String, String[]> columns = new HashMap<String, String[]>(); + for (int i = 0; i < columnFamilies.length; i++) { + columns.put(columnFamilies[i], genWords(random, MIN_NUM_COLS, MAX_NUM_COLS, "col")); + } + IndexWriter writer = new IndexWriter(directory, new IndexWriterConfig(LUCENE_VERSION, new StandardAnalyzer(LUCENE_VERSION))); + int numberOfDocs = random.nextInt(MAX_NUM_OF_DOCS) + 1; + for (int i = 0; i < numberOfDocs; i++) { + writer.addDocuments(generatSuperDoc(random, columns, sampler)); + } + writer.close(); return directory; } @@ -106,37 +117,35 @@ public class RandomSuperQueryTest { return str; } -// private Row generatSuperDoc(Random random, Map<String, String[]> columns, Collection<String> sampler) { -// Row row = new Row().setId(Long.toString(random.nextLong())); -// StringBuilder builder = new StringBuilder(); -// for (String colFam : columns.keySet()) { -// String[] cols = columns.get(colFam); -// for (int i = 0; i < random.nextInt(MAX_NUM_DOCS_PER_COL_FAM); i++) { -// Record record = new Record(); -// record.setFamily(colFam); -// record.setRecordId(Long.toString(random.nextLong())); -// int staringLength = builder.length(); -// for (String column : cols) { -// if (random.nextInt() % MOD_COLS_USED_FOR_SKIPPING == 0) { -// String word = genWord(random, "word"); -// record.addToColumns(new Column(column, word)); -// if (random.nextInt() % MOD_USED_FOR_SAMPLING == 0) { -// builder.append(" +" + colFam + "." + column + ":" + word); -// } -// } -// } -// if (builder.length() != staringLength) { -// builder.append(" nojoin.nojoin "); -// } -// row.addToRecords(record); -// } -// } -// String string = builder.toString().trim(); -// if (!string.isEmpty()) { -// sampler.add(string); -// } -// return row; -// } + private List<Document> generatSuperDoc(Random random, Map<String, String[]> columns, Collection<String> sampler) { + List<Document> docs = new ArrayList<Document>(); + StringBuilder builder = new StringBuilder(); + for (String colFam : columns.keySet()) { + String[] cols = columns.get(colFam); + for (int i = 0; i < random.nextInt(MAX_NUM_DOCS_PER_COL_FAM); i++) { + Document doc = new Document(); + int staringLength = builder.length(); + for (String column : cols) { + if (random.nextInt() % MOD_COLS_USED_FOR_SKIPPING == 0) { + String word = genWord(random, "word"); + doc.add(new StringField(colFam + "." + column, word, Store.YES)); + if (random.nextInt() % MOD_USED_FOR_SAMPLING == 0) { + builder.append(" +" + colFam + "." + column + ":" + word); + } + } + } + if (builder.length() != staringLength) { + builder.append(" nojoin.nojoin "); + } + docs.add(doc); + } + } + String string = builder.toString().trim(); + if (!string.isEmpty()) { + sampler.add(string); + } + return docs; + } private String genWord(Random random, String prefix) { return prefix + random.nextInt(MAX_NUM_OF_WORDS); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ed671149/src/blur-gui/src/main/java/org/apache/blur/gui/HttpJettyServer.java ---------------------------------------------------------------------- diff --git a/src/blur-gui/src/main/java/org/apache/blur/gui/HttpJettyServer.java b/src/blur-gui/src/main/java/org/apache/blur/gui/HttpJettyServer.java index d775bb8..3ef5b5f 100644 --- a/src/blur-gui/src/main/java/org/apache/blur/gui/HttpJettyServer.java +++ b/src/blur-gui/src/main/java/org/apache/blur/gui/HttpJettyServer.java @@ -23,6 +23,10 @@ import java.util.Properties; import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; import org.apache.blur.metrics.BlurMetrics; +import org.apache.thrift.TProcessor; +import org.apache.thrift.protocol.TJSONProtocol; +import org.apache.thrift.protocol.TProtocolFactory; +import org.apache.thrift.server.TServlet; import org.mortbay.jetty.Server; import org.mortbay.jetty.servlet.ServletHolder; import org.mortbay.jetty.webapp.WebAppContext; @@ -53,10 +57,11 @@ public class HttpJettyServer { * location of webapp to serve * @param bm * metrics object for using. + * @param processor * @throws IOException */ - public HttpJettyServer(int bindPort, int port, int baseControllerPort, int baseShardPort, int baseGuiControllerPort, int baseGuiShardPort, String base, BlurMetrics bm) - throws IOException { + public HttpJettyServer(int bindPort, int port, int baseControllerPort, int baseShardPort, int baseGuiControllerPort, int baseGuiShardPort, String base, BlurMetrics bm, + TProcessor processor) throws IOException { server = new Server(port); String logDir = System.getProperty("blur.logs.dir"); @@ -75,6 +80,10 @@ public class HttpJettyServer { context.setWar(warPath); context.setContextPath("/"); context.setParentLoaderPriority(true); + TProtocolFactory protocolFactory = new TJSONProtocol.Factory(); + if (processor != null) { + context.addServlet(new ServletHolder(new TServlet(processor, protocolFactory)), "/json"); + } context.addServlet(new ServletHolder(new LiveMetricsServlet()), "/livemetrics"); context.addServlet(new ServletHolder(new MetricsServlet(bm)), "/metrics"); context.addServlet(new ServletHolder(new LogServlet(blurLogFile)), "/logs");
