Add STOMP connectors: StompClientSource and StompClientSink #514 #856
Conversation
nachinius
changed the title
…at are not present in 2.11
| val Stomp = Seq( | ||
| libraryDependencies ++= Seq( | ||
| // https://mvnrepository.com/artifact/io.vertx/vertx-stomp | ||
| "io.vertx" % "vertx-stomp" % "3.5.1" // ApacheV2/EPL1 |
There was a problem hiding this comment.
Do we have a sense of the dependencies here ie is it bringing in many?
There was a problem hiding this comment.
What’s the thread usage of this dependency in terms of the number of threads required?
There was a problem hiding this comment.
I’ve implemented STOMP before - the protocol is easy. Should we even need this dependency?
There was a problem hiding this comment.
Thanks for your comments @huntc ,
Do we have a sense of the dependencies here ie is it bringing in many?
It brings stuff from vertx like vertx-core
What’s the thread usage of this dependency in terms of the number of threads required?
How would you check it?
I’ve implemented STOMP before - the protocol is easy. Should we even need this dependency?
Good! Congrats! is it in scala or java? will you open source it?
This alpakka connector basically sits on top of the particular stomp protocol. Our main objective is able to expose Stomp streams. Nevertheless, we could implement the stomp protocol in following versions of this connector to drop the depencency on vertx-stomp. The advantage of vertx-stomp is that is complete, and can handle out of the box several versions of stomp and all features.
In short: eventually we don't need this dependency, but for the moment works.
There was a problem hiding this comment.
I have no code for you unfortunately.
Profiling this would be great to get a sense of resource usage along with thread usage.
There was a problem hiding this comment.
Profiling this is not in my tool belt yet. For the sake of awareness, could you point me in the direction on where to start (blog, book, etc)? thks
There was a problem hiding this comment.
Perhaps start with VisualVM.
| * Connects to a STOMP server upon materialization and sends incoming messages to the server. | ||
| * Each materialized sink will create one connection to the server. | ||
| */ | ||
| final class SinkStage(settings: ConnectorSettings) |
There was a problem hiding this comment.
At a min, we should have a Source and Flow. Sink is a convenience on Flow. Please see the guidelines for more info.
There was a problem hiding this comment.
Flows should also permit an opaque element to be carried through them. I think this is also in the guidelines.
There was a problem hiding this comment.
@huntc we do have a Source in SourceStage.scala.
Regarding the Flow, at this stage, it's not clear to me what do you mean. Could you point me the place in the guidelines that talk about it, since I'm unable to find it. Furthermore, and an example of a flow with the opaque element. Neither I was able to find it in the guidelines. Thanks
There was a problem hiding this comment.
Here are the guidelines I referenced: https://github.com/akka/alpakka/blob/master/contributor-advice.md
There was a problem hiding this comment.
Yes, I had read those. I can't find what you mention. Nevertheless, the basic blocks to construct anything, AFAIU is a Source and a Sink. Later we could add flows, that connects those sources and sink, depending in the most common use cases.
Anyways, what kind of Flow are you imaging? A flow in a stream, that has a side effect of sending the whatever is being passed to a stomp server? That is what you meant by opaque element?
There was a problem hiding this comment.
Hopefully, these links will help:
https://github.com/akka/alpakka/blob/master/contributor-advice.md#public-factory-methods
https://github.com/akka/alpakka/blob/master/contributor-advice.md#flows
You generally construct sinks from flows:
https://github.com/akka/alpakka/blob/master/contributor-advice.md#keep-the-code-dry
Some examples, also illustrating the passing through of an element:
/**
* Append a command request to a queue
* @tparam A The type of data to carry through
* @return a flow that takes a command and returns with an acknowledgement
*/
def flow[A]: Flow[CommandRequest[A], CommandReply[A], NotUsed]
/**
* Append a command request to a queue via a sink for convenience
* @tparam A The type of data to carry through
* @return a sink that takes a command and returns with an acknowledgement
*/
def sink[A]: Sink[CommandRequest[A], Future[Done]] =
flow.toMat(Sink.ignore)(Keep.right)
There was a problem hiding this comment.
Oh, I see what you say. For me, the example is key. I had already read the guidelines links a month ago and didn't get this idea. The example is crystal, can it be added to the contributor-advice? (is there any copyright restriction on the example you wrote?)
There was a problem hiding this comment.
I also see the approach you propose is cleaner and more powerful. I would love to have hear this when I outlined the shape and plan for this contribution on feb 26th ! A month ago! (#514 (comment))
For the moment I'm reluctant to switch approaches. Since this works for several use cases that I find useful.
|
Thank you for this effort. |
|
As Vert.x supports the Reactive Streams specification just as Akka Streams do, we should be able to connect the two and make use of connectors implemented for Vert.x. |
|
I learned a bit more about vert.x and understand now that only a few data structures support Reactive Streams, not vert.x as such. This makes integrating with their connectors harder than I hoped it to be. |
|
...or just avoid vertx for stomp? |
|
Any further thoughts on this PR @ennru ? |
|
Not really. While it would be great to have support for STOMP, we don't want to pull in Vert.x to accomplish it. The best would be to find a more (thread-)lightweight client for STOMP. |
|
Sorry, @nachinius if we didn't communicate well enough before you put the effort in implementing a STOMP connector based on Vertx. As it seems they have a good implementation of the protocol itself and there are few other implementations available so it would be great to re-use that part. Closing this now. |
Add connectors to a STOMP server as a stream client or source.
Ref #514
Before finishing the documentation and test for java, I wish this to be reviewed.
As stated in #514 (comment) here we have
the
and both support backpropagation