Hello everyone,
Sorry about the other e-mail (which created a different thread) and it was
missing Alex's email. So I am now replying to all for this one. Hopefully, this
will come under the original e-mail thread.
I also figured it out as I discovered I had to specify the type to be a topic.
This is what I did previously:
./spout test.fanout
This is where I got the NPE. After some trial and error plus Alex's tip from
below, I was able to get it working with this command:
./spout "test.fanout; {node:{type:topic}}"
Alex already raised a JIRA (QPID-4671) for the NPE (see his email below).
Thanks for everybody's help with this one!
Regards,
Jeremy
________________________________
From: Oleksandr Rudyy <[email protected]>
To: Jeremy Wagner <[email protected]>
Sent: Tuesday, March 26, 2013 5:37 PM
Subject: Re: How to create a fanout exchange with the Java broker
Hi Jeremy,
You can make "spout" to publish messages into fanout exchange (without
running into NullPointerException ) by specifying a node type as topic
in your address. For example, the following worked for me:
./examples/messaging/spout -b guest@localhost:5672
"test.fanout;{node:{type:queue}}"
Having set node type in address should stop cpp qpid client from
sending a command "exchange.bound" to java broker on trying to resolve
the address destination. If node type is not set in address the cpp
client sends "exchange.bound" command with both queue and exchange
arguments set to value "test.fanout" in your case. The java broker
cannot find a queue with name "test.fanout" and due to bug in a fanout
exchange a NullPointerException is thrown.
I raised JIRA https://issues.apache.org/jira/browse/QPID-4671 to
address this issue. After fixing this bug it should be possible to
publish message into fanout exchange using spout example with address
having only exchange name in it:
./examples/messaging/spout -b guest@localhost:5672 test.fanout
Kind Regards,
Alex
On 26 March 2013 17:03, Jeremy Wagner <[email protected]> wrote:
> Hello Rob,
>
> I got the NPE from the System.out to the console, but it did not provide
> sufficient information as shown below.
>
> <timestamp> [Client] warning Exception received from broker: internal-error:
> Exception processing command: java.lang.NullPointerException [caused by 1
> \x00:\x00]
> internal-error: Exception processing command: java.lang.NullPointerException
> <timestamp> [Client] warning Ignoring frame while closing connection:
> Frame[BEbe; channel=1; {SessionKnownCompletedBody: commands={ [0,0] }; }]
>
> However, when you brought up the log, I took a look at it and it had the
> full stack trace, which I am posting below. I looked at the code (the
> isBound method) and it is just returning whether if it contains the queue
> name, but it appears that the passed queue parameter was null as the
> containsKey method will throw a NPE if the specified key passed was null
> (see the javadoc:
> http://docs.oracle.com/javase/6/docs/api/java/util/concurrent/ConcurrentHashMap.html#containsKey(java.lang.Object).
> So the queue parameter probably should be checked for null to fix that NPE.
>
> <timestamp> ERROR [IoReceiver - /<ip address>]
> (ServerSessionDelegate.java:119) - Exception processing command
> java.lang.NullPointerException
> at
> java.util.concurrent.ConcurrentHashMap.containsKey(ConcurrentHashMap.java:782)
> at
> org.apache.qpid.server.exchange.FanoutExchange.isBound(FanoutExchange.java:111)
> at
> org.apache.qpid.server.exchange.AbstractExchange.isBound(AbstractExchange.java:182)
> at
> org.apache.qpid.server.transport.ServerSessionDelegate.exchangeBound(ServerSessionDelegate.java:1159)
> at
> org.apache.qpid.server.transport.ServerSessionDelegate.exchangeBound(ServerSessionDelegate.java:74)
> at org.apache.qpid.transport.ExchangeBound.dispatch(ExchangeBound.java:99)
> at
> org.apache.qpid.transport.SessionDelegate.command(SessionDelegate.java:55)
> at
> org.apache.qpid.server.transport.ServerSessionDelegate.command(ServerSessionDelegate.java:98)
> at
> org.apache.qpid.server.transport.ServerSessionDelegate.command(ServerSessionDelegate.java:74)
> at org.apache.qpid.transport.Method.delegate(Method.java:159)
> at org.apache.qpid.transport.Session.received(Session.java:584)
> at org.apache.qpid.transport.Connection.dispatch(Connection.java:421)
> at
> org.apache.qpid.transport.ConnectionDelegate.handle(ConnectionDelegate.java:64)
> at
> org.apache.qpid.transport.ConnectionDelegate.handle(ConnectionDelegate.java:40)
> at
> org.apache.qpid.transport.MethodDelegate.exchangeBound(MethodDelegate.java:203)
> at org.apache.qpid.transport.ExchangeBound.dispatch(ExchangeBound.java:99)
> at
> org.apache.qpid.transport.ConnectionDelegate.command(ConnectionDelegate.java:54)
> at
> org.apache.qpid.transport.ConnectionDelegate.command(ConnectionDelegate.java:40)
> at org.apache.qpid.transport.Method.delegate(Method.java:159)
> at org.apache.qpid.transport.Connection.received(Connection.java:376)
> at
> org.apache.qpid.server.transport.ServerConnection.received(ServerConnection.java:212)
> at
> org.apache.qpid.server.transport.ServerConnection.received(ServerConnection.java:56)
> at org.apache.qpid.transport.network.Assembler.emit(Assembler.java:97)
> at org.apache.qpid.transport.network.Assembler.assemble(Assembler.java:198)
> at org.apache.qpid.transport.network.Assembler.frame(Assembler.java:131)
> at org.apache.qpid.transport.network.Frame.delegate(Frame.java:128)
> at org.apache.qpid.transport.network.Assembler.received(Assembler.java:102)
> at org.apache.qpid.transport.network.Assembler.received(Assembler.java:44)
> at
> org.apache.qpid.transport.network.InputHandler.next(InputHandler.java:189)
> at
> org.apache.qpid.transport.network.InputHandler.received(InputHandler.java:105)
> at
> org.apache.qpid.server.protocol.ProtocolEngine_0_10.received(ProtocolEngine_0_10.java:109)
> at
> org.apache.qpid.server.protocol.ProtocolEngine_0_10.received(ProtocolEngine_0_10.java:42)
> at
> org.apache.qpid.server.protocol.MultiVersionProtocolEngine.received(MultiVersionProtocolEngine.java:118)
> at
> org.apache.qpid.server.protocol.MultiVersionProtocolEngine.received(MultiVersionProtocolEngine.java:37)
> at org.apache.qpid.transport.network.io.IoReceiver.run(IoReceiver.java:152)
> at java.lang.Thread.run(Thread.java:662)
> <timestamp> INFO [IoReceiver - /<ip address>] (Log4jMessageLogger.java:73)
> - [con:3(<ip address>] [con:3(<ip address>] CHN-1003 : Close
> <timestamp> INFO [IoReceiver - /<ip address>] (Log4jMessageLogger.java:73)
> - [con:3(/<ip address>)] [con:3(/<ip address>)] CON-1002 : Close
>
> So it appears that I may be incorrectly passing in the wrong address as I am
> giving spout the exchange name to send a message? It is my understanding
> that I'm supposed to send a message directly to the exchange so that the
> queues bounded to that exchange will each receive a copy of the message.
> Please correct me if I'm mistaken. Thanks.
>
> Regards,
> Jeremy
> ----- Original Message -----
> From: Rob Godfrey <[email protected]>
> To: [email protected]
> Cc: "[email protected]" <[email protected]>
> Sent: Tuesday, March 26, 2013 9:22 AM
> Subject: Re: How to create a fanout exchange with the Java broker
>
> Hi Jeremy,
>
> when you say you got an NPE from the Java Broker, do you mean that there
> was an NPE in the log or System.out from the broker? If so can you post
> the stack trace here - we should definitely not do that no matter what
> address you are using.
>
> thanks,
> Rob
>
> On 26 March 2013 17:11, Jeremy Wagner <[email protected]> wrote:
>
>> Hello Alex,
>>
>> Thank you for your help! I was able to get it configured correctly. I
>> created three queues to bind to the exchange. I then connected separate
>> listeners (using drain) to each queue. I attempted to send a message to
>> the
>> exchange using spout, but I got a NullPointerException from the Java
>> broker. I believe I am not specifying the address to the fanout exchange
>> correctly? Using the example that you gave me, what would be the correct
>> address for the test.fanout exchange for spout to be able to connect to
>> send a message? Thank you!
>>
>> Jeremy
>>
>>
