Github user tmalaska commented on a diff in the pull request:
https://github.com/apache/spark/pull/566#discussion_r14040514
--- Diff:
external/flume/src/main/scala/org/apache/spark/streaming/flume/FlumeInputDStream.scala
---
@@ -134,22 +144,64 @@ private[streaming]
class FlumeReceiver(
host: String,
port: Int,
- storageLevel: StorageLevel
+ storageLevel: StorageLevel,
+ enableDecompression: Boolean
) extends Receiver[SparkFlumeEvent](storageLevel) with Logging {
lazy val responder = new SpecificResponder(
classOf[AvroSourceProtocol], new FlumeEventServer(this))
- lazy val server = new NettyServer(responder, new InetSocketAddress(host,
port))
+ var server: NettyServer = null
+
+ private def initServer() = {
+ if (enableDecompression) {
+ val channelFactory = new NioServerSocketChannelFactory
+ (Executors.newCachedThreadPool(), Executors.newCachedThreadPool());
+ val channelPipelieFactory = new CompressionChannelPipelineFactory()
+
+ new NettyServer(
+ responder,
+ new InetSocketAddress(host, port),
+ channelFactory,
+ channelPipelieFactory,
+ null)
+ } else {
+ new NettyServer(responder, new InetSocketAddress(host, port))
+ }
+ }
def onStart() {
- server.start()
+ synchronized {
+ if (server == null) {
+ server = initServer()
+ server.start()
+ } else {
+ logWarning("Flume receiver being asked to start more then once
with out close")
+ }
+ }
logInfo("Flume receiver started")
}
def onStop() {
- server.close()
+ synchronized {
+ if (server != null) {
+ server.close()
+ server = null
+ }
+ }
logInfo("Flume receiver stopped")
}
override def preferredLocation = Some(host)
}
+
+private[streaming]
+class CompressionChannelPipelineFactory extends ChannelPipelineFactory {
+
+ def getPipeline() = {
--- End diff --
Cool will do before the weekend is done. Thanks
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
