[jira] [Commented] (GEARPUMP-303) add a RabbitMQ sink to integrate with gearpump

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/GEARPUMP-303?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15999259#comment-15999259
 ] 

ASF GitHub Bot commented on GEARPUMP-303:
-

GitHub user yanghua opened a pull request:

https://github.com/apache/incubator-gearpump/pull/180

[GEARPUMP-303] add a RabbitMQ sink to integrate with gearpump

Be sure to do all of the following to help us incorporate your contribution
quickly and easily:

 - [ ] Make sure the commit message is formatted like:
   `[GEARPUMP-] Meaningful description of pull request` 
 - [ ] Make sure tests pass via `sbt clean test`.
 - [ ] Make sure old documentation affected by the pull request has been 
updated and new documentation added for new functionality. 



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/yanghua/incubator-gearpump dev

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/incubator-gearpump/pull/180.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #180






> add a RabbitMQ sink to integrate with gearpump
> --
>
> Key: GEARPUMP-303
> URL: https://issues.apache.org/jira/browse/GEARPUMP-303
> Project: Apache Gearpump
>  Issue Type: Task
>  Components: connectors
>Reporter: yanghua
>Assignee: yanghua
>




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

Re: We got MessageLossException from task Some(TaskId(0,0)), replaying application...

[Karol Brejna]

Ramin,
Is there a chance for you to share the code?

What is your datasource?

Regards,
Karol

W dniu pt., 5.05.2017 o 18:06 Ramin Etemaadi 
napisał(a):

> Hi team,
>
> Thanks for your great job on Gearpump project.
>
> After upgrading from 0.8.1 to 0.8.3 (and adjusting changing interfaces), I
> got this error in the logs:
> "We got MessageLossException from task Some(TaskId(0,0)), replaying
> application..."
>
> This causes the failure of the pipeline.
>
> I searched the internet and I couldn't find any clue, can you help me with
> a indicator that what could be a stating point to debug this issue?
>
> Regards,
> Ramin
>

We got MessageLossException from task Some(TaskId(0,0)), replaying application...

[Ramin Etemaadi]

Hi team,

Thanks for your great job on Gearpump project.

After upgrading from 0.8.1 to 0.8.3 (and adjusting changing interfaces), I
got this error in the logs:
"We got MessageLossException from task Some(TaskId(0,0)), replaying
application..."

This causes the failure of the pipeline.

I searched the internet and I couldn't find any clue, can you help me with
a indicator that what could be a stating point to debug this issue?

Regards,
Ramin

[GitHub] incubator-gearpump issue #179: test

[codecov-io]

Github user codecov-io commented on the issue:

https://github.com/apache/incubator-gearpump/pull/179
  
# 
[Codecov](https://codecov.io/gh/apache/incubator-gearpump/pull/179?src=pr=h1)
 Report
> Merging 
[#179](https://codecov.io/gh/apache/incubator-gearpump/pull/179?src=pr=desc) 
into 
[master](https://codecov.io/gh/apache/incubator-gearpump/commit/9798863753dac06077ce23bef5f451bf7fdca619?src=pr=desc)
 will **increase** coverage by `0.21%`.
> The diff coverage is `n/a`.



```diff
@@Coverage Diff @@
##   master #179  +/-   ##
==
+ Coverage   71.63%   71.85%   +0.21% 
==
  Files 191  191  
  Lines6086 6086  
  Branches  527  527  
==
+ Hits 4360 4373  +13 
+ Misses   1726 1713  -13
```




---
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.
---

[GitHub] incubator-gearpump pull request #179: test

[karol-brejna-i]

GitHub user karol-brejna-i opened a pull request:

https://github.com/apache/incubator-gearpump/pull/179

test

Be sure to do all of the following to help us incorporate your contribution
quickly and easily:

 - [ ] Make sure the commit message is formatted like:
   `[GEARPUMP-] Meaningful description of pull request` 
 - [ ] Make sure tests pass via `sbt clean test`.
 - [ ] Make sure old documentation affected by the pull request has been 
updated and new documentation added for new functionality. 



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/karol-brejna-i/incubator-gearpump 
GEARPUMP-297-fix-docker-links

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/incubator-gearpump/pull/179.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #179






---
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.
---

[GitHub] incubator-gearpump pull request #179: test

[karol-brejna-i]

Github user karol-brejna-i closed the pull request at:

https://github.com/apache/incubator-gearpump/pull/179


---
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.
---

[GitHub] incubator-gearpump pull request #178: [GEARPUMP-303] add a RabbitMQ sink to ...

[huafengw]

Github user huafengw commented on a diff in the pull request:

https://github.com/apache/incubator-gearpump/pull/178#discussion_r114961959
  
--- Diff: 
external/rabbitmq/src/main/scala/org/apache/gearpump/external/rabbitmq/RMQSink.scala
 ---
@@ -0,0 +1,177 @@
+/*
+ * 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.
+ */
+
+package org.apache.gearpump.external.rabbitmq
+
+import org.apache.gearpump.Message
+import org.apache.gearpump.cluster.UserConfig
+import org.apache.gearpump.streaming.sink.DataSink
+import org.apache.gearpump.streaming.task.TaskContext
+import com.rabbitmq.client.Channel
+import com.rabbitmq.client.{Connection, ConnectionFactory}
+
+class RMQSink(userConfig: UserConfig) extends DataSink{
+
+  var connection: Connection = null
+  var channel: Channel = null
+  var queueName: String = null
+
+  override def open(context: TaskContext): Unit = {
+val factory : ConnectionFactory = 
RMQSink.getConnectionFactory(userConfig)
+connection = factory.newConnection
+channel = connection.createChannel
+if (channel == null) {
+  throw new RuntimeException("None of RabbitMQ channels are 
available.")
+}
+setupQueue()
+  }
+
+  override def write(message: Message): Unit = {
+publish(message.msg)
+  }
+
+  override def close(): Unit = {
+channel.close()
+connection.close()
+  }
+
+  protected def setupQueue(): Unit = {
+val queue = RMQSink.getQueueName(userConfig)
+if (!queue.nonEmpty) {
+  throw new RuntimeException("can not get a RabbitMQ queue name")
+}
+
+queueName = queue.get
+channel.queueDeclare(queue.get, false, false, false, null)
+  }
+
+  def publish(msg: Any): Unit = {
+msg match {
+  case seq: Seq[Any] =>
+seq.foreach(publish)
+  case str: String => {
+channel.basicPublish("", queueName, null, 
msg.asInstanceOf[String].getBytes)
+  }
+  case byteArray: Array[Byte@unchecked] => {
+channel.basicPublish("", queueName, null, byteArray)
+  }
+  case _ => {
+
+  }
+}
+  }
+
+}
+
+object RMQSink {
+
+  val RMQSINK = "rmqsink"
+  val QUEUE_NAME = "rabbitmq.queue.name"
+  val SERVER_HOST = "rabbitmq.connection.host"
+  val SERVER_PORT = "rabbitmq.connection.port"
+  val CONNECTION_URI = "rabbitmq.connection.uri"
+  val VIRTUAL_HOST = "rabbitmq.virtualhost"
+  val AUTH_USERNAME = "rabbitmq.auth.username"
+  val AUTH_PASSWORD = "rabbitmq.auth.password"
+  val AUTOMATIC_RECOVERY = "rabbitmq.automatic.recovery"
+  val CONNECTION_TIMEOUT = "rabbitmq.connection.timeout"
+  val NETWORK_RECOVERY_INTERVAL = "rabbitmq.network.recovery.interval"
+  val REQUESTED_HEARTBEAT = "rabbitmq.requested.heartbeat"
+  val TOPOLOGY_RECOVERY_ENABLED = "rabbitmq.topology.recoveryenabled"
+  val REQUESTED_CHANNEL_MAX = "rabbitmq.channel.max"
+  val REQUESTED_FRAME_MAX = "rabbitmq.frame.max"
+
+  def getConnectionFactory(userConfig : UserConfig): ConnectionFactory = {
+val factory : ConnectionFactory = new ConnectionFactory;
--- End diff --

don't need comma here


---
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.
---

[GitHub] incubator-gearpump pull request #178: [GEARPUMP-303] add a RabbitMQ sink to ...

[huafengw]

Github user huafengw commented on a diff in the pull request:

https://github.com/apache/incubator-gearpump/pull/178#discussion_r114962623
  
--- Diff: external/rabbitmq/README.md ---
@@ -0,0 +1,21 @@
+# Gearpump RabbitMQ
+
+Gearpump integration for [RabbitMQ](https://www.rabbitmq.com/)
+
+## Usage
+
+The message type that RMQSink is able to handle including:
+
+ 1. String
+ 2. Array[Byte]
+ 3. Sequence of type 1 and 2
+
+Suppose there is a DataSource Task will output above-mentioned messages, 
you can write a simple application then:
+
+```scala
+val sink = new RMQSink(UserConfig.empty, "$tableName")
+val sinkProcessor = DataSinkProcessor(sink, "$sinkNum")
+val split = Processor[DataSource]("$splitNum")
+val computation = split ~> sinkProcessor
+val application = StreamApplication("RabbitMQ", Graph(computation), 
UserConfig.empty)
+```
--- End diff --

Probably we also need some configuration doc here, like  
`rabbitmq.connection.host ` etc.


---
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.
---

[GitHub] incubator-gearpump pull request #178: [GEARPUMP-303] add a RabbitMQ sink to ...

[huafengw]

Github user huafengw commented on a diff in the pull request:

https://github.com/apache/incubator-gearpump/pull/178#discussion_r114962293
  
--- Diff: 
external/rabbitmq/src/test/scala/org/apache/gearpump/external/rabbitmq/RabbitmqSinkSpec.scala
 ---
@@ -0,0 +1,51 @@
+/*
+ * 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.
+ */
+package org.apache.gearpump.external.rabbitmq
+
+import org.apache.gearpump.Message
+import org.apache.gearpump.cluster.UserConfig
+import org.apache.gearpump.streaming.task.TaskContext
+import org.mockito.Mockito._
+import org.scalatest.mock.MockitoSugar
+import org.scalatest.prop.PropertyChecks
+import org.scalatest.{Matchers, PropSpec}
+
+class RabbitmqSinkSpec extends PropSpec with PropertyChecks with Matchers 
with MockitoSugar {
+
+  property("HBaseSink should insert a row successfully") {
--- End diff --

The assert message is not correct


---
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.
---

[GitHub] incubator-gearpump pull request #178: [GEARPUMP-303] add a RabbitMQ sink to ...

[yanghua]

GitHub user yanghua opened a pull request:

https://github.com/apache/incubator-gearpump/pull/178

[GEARPUMP-303] add a RabbitMQ sink to integrate with gearpump



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/yanghua/incubator-gearpump master

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/incubator-gearpump/pull/178.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #178


commit 32b18443d186937b272f8143041ff999fee4ccce
Author: vinoyang 
Date:   2017-05-05T08:10:17Z

[GEARPUMP-303] add a RabbitMQ sink to integrate with gearpump




---
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.
---

Re: GSoC 2017 student projects announced

[Buddhi Ayesha Rathnayaka]

Yeah, I checked and sent a message to you through Slack.

Thanks & Regards,


On 5 May 2017 at 15:18, Manu Zhang  wrote:

> Buddhi,
>
> I've sent you an invitation to our slack channel, which would allow for
> faster communication.
> Could you check ?
>
> Thanks,
> Manu
>
> On Fri, May 5, 2017 at 4:40 PM Karol Brejna 
> wrote:
>
> > Buddhi,
> >
> > Congrats and good luck!
> > I'll keep my fingers crossed ;-)
> >
> > Regards,
> > Karol
> >
> > On Fri, May 5, 2017 at 7:43 AM, Buddhi Ayesha Rathnayaka
> >  wrote:
> > > I am thrilled to have this great opportunity. First, I want to thank
> Manu
> > > Zhang for helping me throughout the GSoC proposal submission process.
> It
> > > was the good communication between ourselves that helped me get beyond
> > the
> > > proposal submission process. I hope to continue that until I
> successfully
> > > complete this project.
> > >
> > > For everyone, I want to say Hi!
> > >
> > > I am a final year undergraduate of Computer Science & Engineering,
> > > University of Moratuwa, Sri Lanka. I have experience on working in
> > > open-source java projects. I usually spend my leisure time in
> innovative
> > > projects and present them in hackathons.
> > >
> > > Thanks & Regards,
> > >
> > > On 5 May 2017 at 06:51, Manu Zhang  wrote:
> > >
> > >> Hi All,
> > >>
> > >> The GSoC 2017 student projects have been announced.
> > >> Buddhi Ayesha is the student to work on our Add SQL Support
> > >> 
> > project
> > >> and I'm the mentor.
> > >>
> > >> Congratulations, Buddhi !!!
> > >> I've assigned to GEARPUMP-217
> > >>  to you. Could
> you
> > >> introduce yourself to everyone ?
> > >>
> > >> Thanks,
> > >> Manu
> > >>
> > >>
> > >>
> > >
> > >
> > > --
> > >
> > > *H.K.D. Buddhi Ayesha Rathnayaka*Undergraduate BSc. Eng.(Hons.)
> > > Department of Computer Science & Engineering,
> > > University of Moratuwa.
> > > *Mobile: 0716943927*
> >
>



-- 

*H.K.D. Buddhi Ayesha Rathnayaka*Undergraduate BSc. Eng.(Hons.)
Department of Computer Science & Engineering,
University of Moratuwa.
*Mobile: 0716943927*

Re: GSoC 2017 student projects announced

[Manu Zhang]

Buddhi,

I've sent you an invitation to our slack channel, which would allow for
faster communication.
Could you check ?

Thanks,
Manu

On Fri, May 5, 2017 at 4:40 PM Karol Brejna  wrote:

> Buddhi,
>
> Congrats and good luck!
> I'll keep my fingers crossed ;-)
>
> Regards,
> Karol
>
> On Fri, May 5, 2017 at 7:43 AM, Buddhi Ayesha Rathnayaka
>  wrote:
> > I am thrilled to have this great opportunity. First, I want to thank Manu
> > Zhang for helping me throughout the GSoC proposal submission process. It
> > was the good communication between ourselves that helped me get beyond
> the
> > proposal submission process. I hope to continue that until I successfully
> > complete this project.
> >
> > For everyone, I want to say Hi!
> >
> > I am a final year undergraduate of Computer Science & Engineering,
> > University of Moratuwa, Sri Lanka. I have experience on working in
> > open-source java projects. I usually spend my leisure time in innovative
> > projects and present them in hackathons.
> >
> > Thanks & Regards,
> >
> > On 5 May 2017 at 06:51, Manu Zhang  wrote:
> >
> >> Hi All,
> >>
> >> The GSoC 2017 student projects have been announced.
> >> Buddhi Ayesha is the student to work on our Add SQL Support
> >> 
> project
> >> and I'm the mentor.
> >>
> >> Congratulations, Buddhi !!!
> >> I've assigned to GEARPUMP-217
> >>  to you. Could you
> >> introduce yourself to everyone ?
> >>
> >> Thanks,
> >> Manu
> >>
> >>
> >>
> >
> >
> > --
> >
> > *H.K.D. Buddhi Ayesha Rathnayaka*Undergraduate BSc. Eng.(Hons.)
> > Department of Computer Science & Engineering,
> > University of Moratuwa.
> > *Mobile: 0716943927*
>

[jira] [Commented] (GEARPUMP-303) add a RabbitMQ sink to integrate with gearpump

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/GEARPUMP-303?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15998007#comment-15998007
 ] 

ASF GitHub Bot commented on GEARPUMP-303:
-

Github user huafengw commented on a diff in the pull request:

https://github.com/apache/incubator-gearpump/pull/178#discussion_r114957864
  
--- Diff: 
external/rabbitmq/src/main/scala/org/apache/gearpump/external/rabbitmq/RMQSink.scala
 ---
@@ -0,0 +1,177 @@
+/*
+ * 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.
+ */
+
+package org.apache.gearpump.external.rabbitmq
+
+import org.apache.gearpump.Message
+import org.apache.gearpump.cluster.UserConfig
+import org.apache.gearpump.streaming.sink.DataSink
+import org.apache.gearpump.streaming.task.TaskContext
+import com.rabbitmq.client.Channel
+import com.rabbitmq.client.{Connection, ConnectionFactory}
+
+class RMQSink(userConfig: UserConfig) extends DataSink{
+
+  var connection: Connection = null
+  var channel: Channel = null
+  var queueName: String = null
+
+  override def open(context: TaskContext): Unit = {
+val factory : ConnectionFactory = 
RMQSink.getConnectionFactory(userConfig)
+connection = factory.newConnection
+channel = connection.createChannel
+if (channel == null) {
+  throw new RuntimeException("None of RabbitMQ channels are 
available.")
+}
+setupQueue()
+  }
+
+  override def write(message: Message): Unit = {
+publish(message.msg)
+  }
+
+  override def close(): Unit = {
+channel.close()
+connection.close()
+  }
+
+  protected def setupQueue(): Unit = {
+val queue = RMQSink.getQueueName(userConfig)
+if (!queue.nonEmpty) {
+  throw new RuntimeException("can not get a RabbitMQ queue name")
+}
+
+queueName = queue.get
+channel.queueDeclare(queue.get, false, false, false, null)
+  }
+
+  def publish(msg: Any): Unit = {
+msg match {
+  case seq: Seq[Any] =>
+seq.foreach(publish)
+  case str: String => {
+channel.basicPublish("", queueName, null, 
msg.asInstanceOf[String].getBytes)
+  }
+  case byteArray: Array[Byte@unchecked] => {
--- End diff --

is `unchecked` annotation needed here?


> add a RabbitMQ sink to integrate with gearpump
> --
>
> Key: GEARPUMP-303
> URL: https://issues.apache.org/jira/browse/GEARPUMP-303
> Project: Apache Gearpump
>  Issue Type: Task
>  Components: connectors
>Reporter: yanghua
>Assignee: yanghua
>




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (GEARPUMP-303) add a RabbitMQ sink to integrate with gearpump

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/GEARPUMP-303?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15998006#comment-15998006
 ] 

ASF GitHub Bot commented on GEARPUMP-303:
-

Github user huafengw commented on a diff in the pull request:

https://github.com/apache/incubator-gearpump/pull/178#discussion_r114962623
  
--- Diff: external/rabbitmq/README.md ---
@@ -0,0 +1,21 @@
+# Gearpump RabbitMQ
+
+Gearpump integration for [RabbitMQ](https://www.rabbitmq.com/)
+
+## Usage
+
+The message type that RMQSink is able to handle including:
+
+ 1. String
+ 2. Array[Byte]
+ 3. Sequence of type 1 and 2
+
+Suppose there is a DataSource Task will output above-mentioned messages, 
you can write a simple application then:
+
+```scala
+val sink = new RMQSink(UserConfig.empty, "$tableName")
+val sinkProcessor = DataSinkProcessor(sink, "$sinkNum")
+val split = Processor[DataSource]("$splitNum")
+val computation = split ~> sinkProcessor
+val application = StreamApplication("RabbitMQ", Graph(computation), 
UserConfig.empty)
+```
--- End diff --

Probably we also need some configuration doc here, like  
`rabbitmq.connection.host ` etc.


> add a RabbitMQ sink to integrate with gearpump
> --
>
> Key: GEARPUMP-303
> URL: https://issues.apache.org/jira/browse/GEARPUMP-303
> Project: Apache Gearpump
>  Issue Type: Task
>  Components: connectors
>Reporter: yanghua
>Assignee: yanghua
>




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (GEARPUMP-303) add a RabbitMQ sink to integrate with gearpump

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/GEARPUMP-303?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15998008#comment-15998008
 ] 

ASF GitHub Bot commented on GEARPUMP-303:
-

Github user huafengw commented on a diff in the pull request:

https://github.com/apache/incubator-gearpump/pull/178#discussion_r114957906
  
--- Diff: 
external/rabbitmq/src/main/scala/org/apache/gearpump/external/rabbitmq/RMQSink.scala
 ---
@@ -0,0 +1,177 @@
+/*
+ * 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.
+ */
+
+package org.apache.gearpump.external.rabbitmq
+
+import org.apache.gearpump.Message
+import org.apache.gearpump.cluster.UserConfig
+import org.apache.gearpump.streaming.sink.DataSink
+import org.apache.gearpump.streaming.task.TaskContext
+import com.rabbitmq.client.Channel
+import com.rabbitmq.client.{Connection, ConnectionFactory}
+
+class RMQSink(userConfig: UserConfig) extends DataSink{
+
+  var connection: Connection = null
+  var channel: Channel = null
+  var queueName: String = null
+
+  override def open(context: TaskContext): Unit = {
+val factory : ConnectionFactory = 
RMQSink.getConnectionFactory(userConfig)
+connection = factory.newConnection
+channel = connection.createChannel
+if (channel == null) {
+  throw new RuntimeException("None of RabbitMQ channels are 
available.")
+}
+setupQueue()
+  }
+
+  override def write(message: Message): Unit = {
+publish(message.msg)
+  }
+
+  override def close(): Unit = {
+channel.close()
+connection.close()
+  }
+
+  protected def setupQueue(): Unit = {
+val queue = RMQSink.getQueueName(userConfig)
+if (!queue.nonEmpty) {
--- End diff --

maybe `queue.isEmpty` is better


> add a RabbitMQ sink to integrate with gearpump
> --
>
> Key: GEARPUMP-303
> URL: https://issues.apache.org/jira/browse/GEARPUMP-303
> Project: Apache Gearpump
>  Issue Type: Task
>  Components: connectors
>Reporter: yanghua
>Assignee: yanghua
>




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (GEARPUMP-303) add a RabbitMQ sink to integrate with gearpump

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/GEARPUMP-303?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15998010#comment-15998010
 ] 

ASF GitHub Bot commented on GEARPUMP-303:
-

Github user huafengw commented on a diff in the pull request:

https://github.com/apache/incubator-gearpump/pull/178#discussion_r114961959
  
--- Diff: 
external/rabbitmq/src/main/scala/org/apache/gearpump/external/rabbitmq/RMQSink.scala
 ---
@@ -0,0 +1,177 @@
+/*
+ * 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.
+ */
+
+package org.apache.gearpump.external.rabbitmq
+
+import org.apache.gearpump.Message
+import org.apache.gearpump.cluster.UserConfig
+import org.apache.gearpump.streaming.sink.DataSink
+import org.apache.gearpump.streaming.task.TaskContext
+import com.rabbitmq.client.Channel
+import com.rabbitmq.client.{Connection, ConnectionFactory}
+
+class RMQSink(userConfig: UserConfig) extends DataSink{
+
+  var connection: Connection = null
+  var channel: Channel = null
+  var queueName: String = null
+
+  override def open(context: TaskContext): Unit = {
+val factory : ConnectionFactory = 
RMQSink.getConnectionFactory(userConfig)
+connection = factory.newConnection
+channel = connection.createChannel
+if (channel == null) {
+  throw new RuntimeException("None of RabbitMQ channels are 
available.")
+}
+setupQueue()
+  }
+
+  override def write(message: Message): Unit = {
+publish(message.msg)
+  }
+
+  override def close(): Unit = {
+channel.close()
+connection.close()
+  }
+
+  protected def setupQueue(): Unit = {
+val queue = RMQSink.getQueueName(userConfig)
+if (!queue.nonEmpty) {
+  throw new RuntimeException("can not get a RabbitMQ queue name")
+}
+
+queueName = queue.get
+channel.queueDeclare(queue.get, false, false, false, null)
+  }
+
+  def publish(msg: Any): Unit = {
+msg match {
+  case seq: Seq[Any] =>
+seq.foreach(publish)
+  case str: String => {
+channel.basicPublish("", queueName, null, 
msg.asInstanceOf[String].getBytes)
+  }
+  case byteArray: Array[Byte@unchecked] => {
+channel.basicPublish("", queueName, null, byteArray)
+  }
+  case _ => {
+
+  }
+}
+  }
+
+}
+
+object RMQSink {
+
+  val RMQSINK = "rmqsink"
+  val QUEUE_NAME = "rabbitmq.queue.name"
+  val SERVER_HOST = "rabbitmq.connection.host"
+  val SERVER_PORT = "rabbitmq.connection.port"
+  val CONNECTION_URI = "rabbitmq.connection.uri"
+  val VIRTUAL_HOST = "rabbitmq.virtualhost"
+  val AUTH_USERNAME = "rabbitmq.auth.username"
+  val AUTH_PASSWORD = "rabbitmq.auth.password"
+  val AUTOMATIC_RECOVERY = "rabbitmq.automatic.recovery"
+  val CONNECTION_TIMEOUT = "rabbitmq.connection.timeout"
+  val NETWORK_RECOVERY_INTERVAL = "rabbitmq.network.recovery.interval"
+  val REQUESTED_HEARTBEAT = "rabbitmq.requested.heartbeat"
+  val TOPOLOGY_RECOVERY_ENABLED = "rabbitmq.topology.recoveryenabled"
+  val REQUESTED_CHANNEL_MAX = "rabbitmq.channel.max"
+  val REQUESTED_FRAME_MAX = "rabbitmq.frame.max"
+
+  def getConnectionFactory(userConfig : UserConfig): ConnectionFactory = {
+val factory : ConnectionFactory = new ConnectionFactory;
--- End diff --

don't need comma here


> add a RabbitMQ sink to integrate with gearpump
> --
>
> Key: GEARPUMP-303
> URL: https://issues.apache.org/jira/browse/GEARPUMP-303
> Project: Apache Gearpump
>  Issue Type: Task
>  Components: connectors
>Reporter: yanghua
>Assignee: yanghua
>




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (GEARPUMP-303) add a RabbitMQ sink to integrate with gearpump

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/GEARPUMP-303?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15998009#comment-15998009
 ] 

ASF GitHub Bot commented on GEARPUMP-303:
-

Github user huafengw commented on a diff in the pull request:

https://github.com/apache/incubator-gearpump/pull/178#discussion_r114962293
  
--- Diff: 
external/rabbitmq/src/test/scala/org/apache/gearpump/external/rabbitmq/RabbitmqSinkSpec.scala
 ---
@@ -0,0 +1,51 @@
+/*
+ * 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.
+ */
+package org.apache.gearpump.external.rabbitmq
+
+import org.apache.gearpump.Message
+import org.apache.gearpump.cluster.UserConfig
+import org.apache.gearpump.streaming.task.TaskContext
+import org.mockito.Mockito._
+import org.scalatest.mock.MockitoSugar
+import org.scalatest.prop.PropertyChecks
+import org.scalatest.{Matchers, PropSpec}
+
+class RabbitmqSinkSpec extends PropSpec with PropertyChecks with Matchers 
with MockitoSugar {
+
+  property("HBaseSink should insert a row successfully") {
--- End diff --

The assert message is not correct


> add a RabbitMQ sink to integrate with gearpump
> --
>
> Key: GEARPUMP-303
> URL: https://issues.apache.org/jira/browse/GEARPUMP-303
> Project: Apache Gearpump
>  Issue Type: Task
>  Components: connectors
>Reporter: yanghua
>Assignee: yanghua
>




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

Re: GSoC 2017 student projects announced

[Karol Brejna]

Buddhi,

Congrats and good luck!
I'll keep my fingers crossed ;-)

Regards,
Karol

On Fri, May 5, 2017 at 7:43 AM, Buddhi Ayesha Rathnayaka
 wrote:
> I am thrilled to have this great opportunity. First, I want to thank Manu
> Zhang for helping me throughout the GSoC proposal submission process. It
> was the good communication between ourselves that helped me get beyond the
> proposal submission process. I hope to continue that until I successfully
> complete this project.
>
> For everyone, I want to say Hi!
>
> I am a final year undergraduate of Computer Science & Engineering,
> University of Moratuwa, Sri Lanka. I have experience on working in
> open-source java projects. I usually spend my leisure time in innovative
> projects and present them in hackathons.
>
> Thanks & Regards,
>
> On 5 May 2017 at 06:51, Manu Zhang  wrote:
>
>> Hi All,
>>
>> The GSoC 2017 student projects have been announced.
>> Buddhi Ayesha is the student to work on our Add SQL Support
>>  project
>> and I'm the mentor.
>>
>> Congratulations, Buddhi !!!
>> I've assigned to GEARPUMP-217
>>  to you. Could you
>> introduce yourself to everyone ?
>>
>> Thanks,
>> Manu
>>
>>
>>
>
>
> --
>
> *H.K.D. Buddhi Ayesha Rathnayaka*Undergraduate BSc. Eng.(Hons.)
> Department of Computer Science & Engineering,
> University of Moratuwa.
> *Mobile: 0716943927*

[jira] [Updated] (GEARPUMP-303) add a RabbitMQ sink to integrate with gearpump

[yanghua (JIRA)]

 [ 
https://issues.apache.org/jira/browse/GEARPUMP-303?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

yanghua updated GEARPUMP-303:
-

hi Karol Brejna,

I think the gearpump currently lack some connector to integrate with other
system or middleware. It is gearpump's limitation. The RabbitMQ is a
widely-used message queue，user processed message can sink to it or just as
a middleware for other system which need a data source.




> add a RabbitMQ sink to integrate with gearpump
> --
>
> Key: GEARPUMP-303
> URL: https://issues.apache.org/jira/browse/GEARPUMP-303
> Project: Apache Gearpump
>  Issue Type: Task
>Reporter: yanghua
>Assignee: yanghua
>




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (GEARPUMP-303) add a RabbitMQ sink to integrate with gearpump

[Karol Brejna (JIRA)]

[ 
https://issues.apache.org/jira/browse/GEARPUMP-303?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15997919#comment-15997919
 ] 

Karol Brejna commented on GEARPUMP-303:
---

Oh, I see you have already prepared a PR! ;-)

Let me look at it...


> add a RabbitMQ sink to integrate with gearpump
> --
>
> Key: GEARPUMP-303
> URL: https://issues.apache.org/jira/browse/GEARPUMP-303
> Project: Apache Gearpump
>  Issue Type: Task
>Reporter: yanghua
>Assignee: yanghua
>




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)