From reviews-return-1062-archive-asf-public=cust-asf.ponee.io@bahir.apache.org Fri Nov 30 00:01:23 2018 Return-Path: X-Original-To: archive-asf-public@cust-asf.ponee.io Delivered-To: archive-asf-public@cust-asf.ponee.io Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by mx-eu-01.ponee.io (Postfix) with SMTP id 4BA5F18066C for ; Fri, 30 Nov 2018 00:01:23 +0100 (CET) Received: (qmail 36032 invoked by uid 500); 29 Nov 2018 23:01:22 -0000 Mailing-List: contact reviews-help@bahir.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: reviews@bahir.apache.org Delivered-To: mailing list reviews@bahir.apache.org Received: (qmail 36018 invoked by uid 99); 29 Nov 2018 23:01:21 -0000 Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org) (140.211.11.23) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 29 Nov 2018 23:01:21 +0000 Received: by git1-us-west.apache.org (ASF Mail Server at git1-us-west.apache.org, from userid 33) id AC12FDFAE2; Thu, 29 Nov 2018 23:01:21 +0000 (UTC) From: lukasz-antoniak To: reviews@bahir.apache.org Reply-To: reviews@bahir.apache.org References: In-Reply-To: Subject: [GitHub] bahir pull request #71: [BAHIR-66] Switch to Java binding for ZeroMQ Content-Type: text/plain Message-Id: <20181129230121.AC12FDFAE2@git1-us-west.apache.org> Date: Thu, 29 Nov 2018 23:01:21 +0000 (UTC) Github user lukasz-antoniak commented on a diff in the pull request: https://github.com/apache/bahir/pull/71#discussion_r237692301 --- Diff: streaming-zeromq/src/main/scala/org/apache/spark/streaming/zeromq/ZeroMQUtils.scala --- @@ -17,147 +17,75 @@ package org.apache.spark.streaming.zeromq +import java.lang.{Iterable => JIterable} +import java.util.{List => JList} + import scala.collection.JavaConverters._ import scala.reflect.ClassTag -import akka.actor.{ActorSystem, Props, SupervisorStrategy} -import akka.util.ByteString -import akka.zeromq.Subscribe - -import org.apache.spark.api.java.function.{Function => JFunction, Function0 => JFunction0} +import org.apache.spark.api.java.function.{Function => JFunction} import org.apache.spark.storage.StorageLevel import org.apache.spark.streaming.StreamingContext -import org.apache.spark.streaming.akka.{ActorReceiver, AkkaUtils} import org.apache.spark.streaming.api.java.{JavaReceiverInputDStream, JavaStreamingContext} import org.apache.spark.streaming.dstream.ReceiverInputDStream object ZeroMQUtils { /** - * Create an input stream that receives messages pushed by a zeromq publisher. - * @param ssc StreamingContext object - * @param publisherUrl Url of remote zeromq publisher - * @param subscribe Topic to subscribe to - * @param bytesToObjects A zeroMQ stream publishes sequence of frames for each topic + * Create an input stream that receives messages pushed by a ZeroMQ publisher. + * @param ssc Streaming context + * @param publisherUrl URL of remote ZeroMQ publisher + * @param connect When positive, connector will try to establish connectivity with remote server. + * Otherwise, it attempts to create and bind local socket. + * @param topics List of topics to subscribe + * @param bytesToObjects ZeroMQ stream publishes sequence of frames for each topic * and each frame has sequence of byte thus it needs the converter --- End diff -- Renamed parameter. Initially I wanted to keep source as close as possible to current master. Provided default message converter supporting text payload, and another variant of method creating input stream. ---