Uploaded image for project: 'AMQ Clients'
  1. AMQ Clients
  2. ENTMQCL-433

[dotnet] Invalid AMQP selector

    XMLWordPrintable

Details

    • Bug
    • Resolution: Won't Do
    • Major
    • None
    • None
    • amqp-dotnet-client
    • None
    • Hide

      1. start amq7 broker
      2. use this reproducer

      using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Transactions;
using Amqp;
using Amqp.Framing;
using Amqp.Types;

namespace test
{
    class Selectors
    {
        public static void RunTest(string[] args)
        {
            Trace.TraceLevel = TraceLevel.Frame;
            Trace.TraceListener = (f, a) => Console.WriteLine(DateTime.Now.ToString("[hh:ss.fff]") + " " + string.Format(f, a));

            Address address = new Address("amqp://10.34.75.113:5672");
            string queue = "jms.queue.test";
            string testName = "Selectors";
            const int nMsgs = 2;

            Connection connection = new Connection(address);
            Session session = new Session(connection);

            ////////////////////////////////////////////////////////////////////////////////////////////////////////

            Source recvSource = new Source();
            recvSource.Address = queue;
            Map filters = new Map();
            filters.Add(new Symbol("filter1"),
                            new DescribedValue(
                                new Symbol("apache.org:selector-filter:string"),
                                "foo/bar"));
            recvSource.FilterSet = filters;

            ReceiverLink receiver = new ReceiverLink(session, "receiver-" + testName, recvSource, (l, a) => { });

            /////////////////////////////////////////////////////////////////////////////////////////////////////////

            SenderLink sender = new SenderLink(session, "sender-" + testName, queue);

            for (int i = 0; i < nMsgs; ++i)
            {
                Message message = new Message("msg" + i);
                message.Properties = new Properties() { GroupId = "abcdefg" };
                message.ApplicationProperties = new ApplicationProperties();
                message.ApplicationProperties["s"] = "testString";
                sender.Send(message);
            }

            for (int i = 0; i < nMsgs; ++i)
            {
                Message message = receiver.Receive(500);
                if (message != null)
                {
                    receiver.Accept(message);
                    Console.WriteLine("Receiver selector: " + message.ApplicationProperties.ToString());
                }
            }

            ////////////////////////////////////////////////////////////////////////////////////////////////////////

            sender.Close();
            receiver.Close();
            session.Close();
            connection.Close();

            Console.ReadLine();
        }
    }
}
      Show
      1. start amq7 broker 2. use this reproducer using System ; using System .Collections.Generic; using System .Linq; using System .Text; using System .Threading.Tasks; using System .Transactions; using Amqp; using Amqp.Framing; using Amqp.Types; namespace test { class Selectors { public static void RunTest( string [] args) { Trace.TraceLevel = TraceLevel.Frame; Trace.TraceListener = (f, a) => Console.WriteLine(DateTime.Now.ToString( "[hh:ss.fff]" ) + " " + string .Format(f, a)); Address address = new Address( "amqp: //10.34.75.113:5672" ); string queue = "jms.queue.test" ; string testName = "Selectors" ; const int nMsgs = 2; Connection connection = new Connection(address); Session session = new Session(connection); //////////////////////////////////////////////////////////////////////////////////////////////////////// Source recvSource = new Source(); recvSource.Address = queue; Map filters = new Map(); filters.Add( new Symbol( "filter1" ), new DescribedValue( new Symbol( "apache.org:selector-filter: string " ), "foo/bar" )); recvSource.FilterSet = filters; ReceiverLink receiver = new ReceiverLink(session, "receiver-" + testName, recvSource, (l, a) => { }); ///////////////////////////////////////////////////////////////////////////////////////////////////////// SenderLink sender = new SenderLink(session, "sender-" + testName, queue); for ( int i = 0; i < nMsgs; ++i) { Message message = new Message( "msg" + i); message.Properties = new Properties() { GroupId = "abcdefg" }; message.ApplicationProperties = new ApplicationProperties(); message.ApplicationProperties[ "s" ] = "testString" ; sender.Send(message); } for ( int i = 0; i < nMsgs; ++i) { Message message = receiver.Receive(500); if (message != null ) { receiver.Accept(message); Console.WriteLine( "Receiver selector: " + message.ApplicationProperties.ToString()); } } //////////////////////////////////////////////////////////////////////////////////////////////////////// sender.Close(); receiver.Close(); session.Close(); connection.Close(); Console.ReadLine(); } } }

    Description

      When I connect receiver with wrong selector, receiver didn't throw exception that receiver is not valid.

      trace client

      [10:27.647] SEND AMQP 0 1.0.0
[10:27.663] SEND (ch=0) open(container-id:c9db084b-df42-46f2-bd6e-80eba01f81e4,host-name:10.34.75.113,max-frame-size:262144,channel-max:256)
[10:27.663] SEND (ch=0) begin(next-outgoing-id:4294967293,incoming-window:2048,outgoing-window:2048,handle-max:4294967295)
[10:27.663] RECV AMQP 0 1 0 0
[10:27.678] SEND (ch=0) attach(name:receiver-Selectors,handle:0,role:True,source:source(address:jms.queue.test,filter:[filter1:foo/bar]),target:target())
[10:27.678] SEND (ch=0) attach(name:sender-Selectors,handle:1,role:False,source:source(),target:target(address:jms.queue.test),initial-delivery-count:0)
[10:27.678] RECV (ch=0) open(container-id:0.0.0.0,host-name:,max-frame-size:4294967295,channel-max:65535,idle-time-out:30000,offered-capabilities:[sole-connection-for-con
tainer,DELAYED_DELIVERY],properties:[product:apache-activemq-artemis,version:1.3.0.amq-700005-redhat-1])
[10:27.694] RECV (ch=0) begin(remote-channel:0,next-outgoing-id:1,incoming-window:2147483647,outgoing-window:2147483647,handle-max:65535)
[10:47.678] RECV (ch=0) attach(name:receiver-Selectors,handle:0,role:False,snd-settle-mode:2,rcv-settle-mode:0,source:source(address:jms.queue.test,durable:0,expiry-polic
y:session-end,timeout:0,dynamic:False,filter:[filter1:foo/bar]),target:target(),incomplete-unsettled:False,initial-delivery-count:0)
[10:47.694] RECV (ch=0) attach(name:sender-Selectors,handle:1,role:True,snd-settle-mode:2,rcv-settle-mode:0,source:source(),target:target(address:jms.queue.test))
[10:47.694] RECV (ch=0) flow(next-in-id:4294967293,in-window:2147483647,next-out-id:1,out-window:2147483647,handle:1,delivery-count:0,link-credit:100)
[10:47.710] SEND (ch=0) transfer(handle:1,delivery-id:0,delivery-tag:00000000,message-format:0,settled:False,batchable:True) payload 55
[10:47.725] RECV (ch=0) disposition(role:True,first:0,last:0,settled:True,state:accepted())
[10:47.725] SEND (ch=0) transfer(handle:1,delivery-id:1,delivery-tag:00000001,message-format:0,settled:False,batchable:True) payload 55
[10:47.741] RECV (ch=0) disposition(role:True,first:1,last:1,settled:True,state:accepted())
[10:47.741] SEND (ch=0) flow(next-in-id:1,in-window:2048,next-out-id:4294967295,out-window:2147483645,handle:0,delivery-count:0,link-credit:200,drain:False)
[10:48.772] SEND (ch=0) detach(handle:1,closed:True)
[10:48.772] RECV (ch=0) detach(handle:1,closed:True)
[10:48.772] SEND (ch=0) detach(handle:0,closed:True)
[10:54.710] SEND (ch=0) empty
[10:58.788] RECV (ch=0) detach(handle:0,closed:True,error:error(condition:amqp:invalid-field,description:foo/bar))
[10:58.788] SEND (ch=0) end()
[10:58.788] RECV (ch=0) end()
[10:58.788] SEND (ch=0) close()
[10:58.803] RECV (ch=0) close()

      trace broker

      [1021793600:0] <- Open{ containerId='c9db084b-df42-46f2-bd6e-80eba01f81e4', hostname='10.34.75.113', maxFrameSize=262144, channelMax=256, idleTimeOut=null, outgoingLocales=null, incomingLocales=null, offeredCapabilities=null, desiredCapabilities=null, properties=null}
[1021793600:0] -> Open{ containerId='0.0.0.0', hostname='', maxFrameSize=4294967295, channelMax=65535, idleTimeOut=30000, outgoingLocales=null, incomingLocales=null, offeredCapabilities=[sole-connection-for-container, DELAYED_DELIVERY], desiredCapabilities=null, properties={product=apache-activemq-artemis, version=1.3.0.amq-700005-redhat-1}}
[1021793600:0] <- Begin{remoteChannel=null, nextOutgoingId=4294967293, incomingWindow=2048, outgoingWindow=2048, handleMax=4294967295, offeredCapabilities=null, desiredCapabilities=null, properties=null}
[1021793600:0] -> Begin{remoteChannel=0, nextOutgoingId=1, incomingWindow=2147483647, outgoingWindow=2147483647, handleMax=65535, offeredCapabilities=null, desiredCapabilities=null, properties=null}
[1021793600:0] <- Attach{name='receiver-Selectors', handle=0, role=RECEIVER, sndSettleMode=MIXED, rcvSettleMode=FIRST, source=Source{address='jms.queue.test', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, distributionMode=null, filter={filter1=org.apache.qpid.proton.codec.DecoderImpl$UnknownDescribedType@19ca33f7}, defaultOutcome=null, outcomes=null, capabilities=null}, target=Target{address='null', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, capabilities=null}, unsettled=null, incompleteUnsettled=false, initialDeliveryCount=null, maxMessageSize=null, offeredCapabilities=null, desiredCapabilities=null, properties=null}
[1021793600:0] <- Attach{name='sender-Selectors', handle=1, role=SENDER, sndSettleMode=MIXED, rcvSettleMode=FIRST, source=Source{address='null', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, distributionMode=null, filter=null, defaultOutcome=null, outcomes=null, capabilities=null}, target=Target{address='jms.queue.test', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, capabilities=null}, unsettled=null, incompleteUnsettled=false, initialDeliveryCount=0, maxMessageSize=null, offeredCapabilities=null, desiredCapabilities=null, properties=null}
[1021793600:0] -> Attach{name='receiver-Selectors', handle=0, role=SENDER, sndSettleMode=MIXED, rcvSettleMode=FIRST, source=Source{address='jms.queue.test', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, distributionMode=null, filter={filter1=org.apache.qpid.proton.codec.DecoderImpl$UnknownDescribedType@19ca33f7}, defaultOutcome=null, outcomes=null, capabilities=null}, target=Target{address='null', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, capabilities=null}, unsettled=null, incompleteUnsettled=false, initialDeliveryCount=0, maxMessageSize=null, offeredCapabilities=null, desiredCapabilities=null, properties=null}
[1021793600:0] -> Attach{name='sender-Selectors', handle=1, role=RECEIVER, sndSettleMode=MIXED, rcvSettleMode=FIRST, source=Source{address='null', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, distributionMode=null, filter=null, defaultOutcome=null, outcomes=null, capabilities=null}, target=Target{address='jms.queue.test', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, capabilities=null}, unsettled=null, incompleteUnsettled=false, initialDeliveryCount=null, maxMessageSize=null, offeredCapabilities=null, desiredCapabilities=null, properties=null}
[1021793600:0] -> Flow{nextIncomingId=4294967293, incomingWindow=2147483647, nextOutgoingId=1, outgoingWindow=2147483647, handle=1, deliveryCount=0, linkCredit=100, available=null, drain=false, echo=false, properties=null}
[1021793600:0] <- Transfer{handle=1, deliveryId=0, deliveryTag=\x00\x00\x00\x00, messageFormat=0, settled=false, more=false, rcvSettleMode=null, state=null, resume=false, aborted=false, batchable=true} (55) "\x00Ss\xc0\x14\x0b@@@@@@@@@@\xa1\x07abcdefg\x00St\xc1\x10\x02\xa1\x01s\xa1\x0atestString\x00Sw\xa1\x04msg0"
[1021793600:0] -> Disposition{role=RECEIVER, first=0, last=0, settled=true, state=Accepted{}, batchable=false}
[1021793600:0] <- Transfer{handle=1, deliveryId=1, deliveryTag=\x00\x00\x00\x01, messageFormat=0, settled=false, more=false, rcvSettleMode=null, state=null, resume=false, aborted=false, batchable=true} (55) "\x00Ss\xc0\x14\x0b@@@@@@@@@@\xa1\x07abcdefg\x00St\xc1\x10\x02\xa1\x01s\xa1\x0atestString\x00Sw\xa1\x04msg1"
[1021793600:0] -> Disposition{role=RECEIVER, first=1, last=1, settled=true, state=Accepted{}, batchable=false}
[1021793600:0] <- Flow{nextIncomingId=1, incomingWindow=2048, nextOutgoingId=4294967295, outgoingWindow=2147483645, handle=0, deliveryCount=0, linkCredit=200, available=null, drain=false, echo=false, properties=null}
10:17:41,601 WARN  [org.proton.plug.handler.impl.ProtonHandlerImpl] null: java.lang.NullPointerException
	at org.apache.activemq.artemis.core.protocol.proton.plug.ProtonSessionIntegrationCallback.onFlowConsumer(ProtonSessionIntegrationCallback.java:126) [artemis-amqp-protocol-1.3.0.amq-700005-redhat-1.jar:]
	at org.proton.plug.context.server.ProtonServerSenderContext.onFlow(ProtonServerSenderContext.java:86) [artemis-proton-plug-1.3.0.amq-700005-redhat-1.jar:1.3.0.amq-700005-redhat-1]
	at org.proton.plug.context.AbstractConnectionContext$LocalListener.onFlow(AbstractConnectionContext.java:316) [artemis-proton-plug-1.3.0.amq-700005-redhat-1.jar:1.3.0.amq-700005-redhat-1]
	at org.proton.plug.handler.Events.dispatch(Events.java:85) [artemis-proton-plug-1.3.0.amq-700005-redhat-1.jar:1.3.0.amq-700005-redhat-1]
	at org.proton.plug.handler.impl.ProtonHandlerImpl.dispatch(ProtonHandlerImpl.java:369) [artemis-proton-plug-1.3.0.amq-700005-redhat-1.jar:1.3.0.amq-700005-redhat-1]
	at org.proton.plug.handler.impl.ProtonHandlerImpl.access$000(ProtonHandlerImpl.java:49) [artemis-proton-plug-1.3.0.amq-700005-redhat-1.jar:1.3.0.amq-700005-redhat-1]
	at org.proton.plug.handler.impl.ProtonHandlerImpl$1.run(ProtonHandlerImpl.java:68) [artemis-proton-plug-1.3.0.amq-700005-redhat-1.jar:1.3.0.amq-700005-redhat-1]
	at org.apache.activemq.artemis.utils.OrderedExecutorFactory$OrderedExecutor$ExecutorTask.run(OrderedExecutorFactory.java:103) [artemis-core-client-1.3.0.amq-700005-redhat-1.jar:1.3.0.amq-700005-redhat-1]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [rt.jar:1.8.0_102]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [rt.jar:1.8.0_102]
	at java.lang.Thread.run(Thread.java:745) [rt.jar:1.8.0_102]

[1021793600:0] <- Detach{handle=1, closed=true, error=null}
[1021793600:0] -> Detach{handle=1, closed=true, error=null}
[1021793600:0] <- Detach{handle=0, closed=true, error=null}
[1021793600:0] <- Empty Frame
[1021793600:0] -> Detach{handle=0, closed=true, error=Error{condition=amqp:invalid-field, description='foo/bar', info=null}}
[1021793600:0] <- End{error=null}
[1021793600:0] -> End{error=null}
[1021793600:0] <- Close{error=null}
[1021793600:0] -> Close{error=null}
10:17:52,653 WARN  [org.apache.activemq.artemis.core.client] AMQ212037: Connection failure has been detected: null [code=REMOTE_DISCONNECT]

      Attachments

        Activity

          People

            crolke@redhat.com Chuck Rolke (Inactive)
            dkornel@redhat.com David Kornel
            Votes:
            0 Vote for this issue
            Watchers:
            3 Start watching this issue

            Dates

              Created:
              Updated:
              Resolved: