You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rh...@apache.org on 2014/03/05 18:55:12 UTC
svn commit: r1574591 - in /qpid/proton/trunk/proton-j/src/main:
java/org/apache/qpid/proton/ java/org/apache/qpid/proton/engine/
java/org/apache/qpid/proton/engine/impl/ resources/
Author: rhs
Date: Wed Mar 5 17:55:11 2014
New Revision: 1574591
URL: http://svn.apache.org/r1574591
Log:
PROTON-527: added events API to proton-j
Added:
qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/Collector.java
qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/Event.java
qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/CollectorImpl.java
qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/EventImpl.java
Modified:
qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/Proton.java
qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/Connection.java
qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/ConnectionImpl.java
qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/EndpointImpl.java
qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/LinkImpl.java
qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/SessionImpl.java
qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java
qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportLink.java
qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportSession.java
qpid/proton/trunk/proton-j/src/main/resources/cengine.py
Modified: qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/Proton.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/Proton.java?rev=1574591&r1=1574590&r2=1574591&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/Proton.java (original)
+++ qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/Proton.java Wed Mar 5 17:55:11 2014
@@ -34,6 +34,7 @@ import org.apache.qpid.proton.codec.Data
import org.apache.qpid.proton.codec.DataFactory;
import org.apache.qpid.proton.driver.Driver;
import org.apache.qpid.proton.driver.DriverFactory;
+import org.apache.qpid.proton.engine.Collector;
import org.apache.qpid.proton.engine.Connection;
import org.apache.qpid.proton.engine.EngineFactory;
import org.apache.qpid.proton.engine.SslDomain;
@@ -44,6 +45,8 @@ import org.apache.qpid.proton.message.Me
import org.apache.qpid.proton.messenger.Messenger;
import org.apache.qpid.proton.messenger.MessengerFactory;
+import org.apache.qpid.proton.engine.impl.CollectorImpl;
+
public final class Proton
{
@@ -75,6 +78,11 @@ public final class Proton
return DEFAULT_IMPLEMENTATION;
}
+ public static Collector collector()
+ {
+ return new CollectorImpl();
+ }
+
public static Connection connection()
{
return ENGINE_FACTORY.createConnection();
Added: qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/Collector.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/Collector.java?rev=1574591&view=auto
==============================================================================
--- qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/Collector.java (added)
+++ qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/Collector.java Wed Mar 5 17:55:11 2014
@@ -0,0 +1,36 @@
+/*
+ *
+ * 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.qpid.proton.engine;
+
+
+/**
+ * Collector
+ *
+ */
+
+public interface Collector
+{
+
+ Event peek();
+
+ void pop();
+
+}
Modified: qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/Connection.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/Connection.java?rev=1574591&r1=1574590&r2=1574591&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/Connection.java (original)
+++ qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/Connection.java Wed Mar 5 17:55:11 2014
@@ -35,6 +35,7 @@ import org.apache.qpid.proton.amqp.Symbo
*/
public interface Connection extends Endpoint
{
+
/**
* Returns a newly created session
*
@@ -99,4 +100,7 @@ public interface Connection extends Endp
Object getContext();
void setContext(Object context);
+
+ void collect(Collector collector);
+
}
Added: qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/Event.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/Event.java?rev=1574591&view=auto
==============================================================================
--- qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/Event.java (added)
+++ qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/Event.java Wed Mar 5 17:55:11 2014
@@ -0,0 +1,53 @@
+/*
+ *
+ * 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.qpid.proton.engine;
+
+
+/**
+ * Event
+ *
+ */
+
+public interface Event
+{
+
+ public enum Type {
+ CONNECTION_STATE,
+ SESSION_STATE,
+ LINK_STATE,
+ LINK_FLOW,
+ DELIVERY,
+ TRANSPORT
+ }
+
+ Type getType();
+
+ Connection getConnection();
+
+ Session getSession();
+
+ Link getLink();
+
+ Delivery getDelivery();
+
+ Transport getTransport();
+
+}
Added: qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/CollectorImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/CollectorImpl.java?rev=1574591&view=auto
==============================================================================
--- qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/CollectorImpl.java (added)
+++ qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/CollectorImpl.java Wed Mar 5 17:55:11 2014
@@ -0,0 +1,60 @@
+/*
+ *
+ * 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.qpid.proton.engine.impl;
+
+import org.apache.qpid.proton.engine.Collector;
+import org.apache.qpid.proton.engine.Event;
+
+import java.util.LinkedList;
+import java.util.Queue;
+
+
+/**
+ * CollectorImpl
+ *
+ */
+
+public class CollectorImpl implements Collector
+{
+
+ private Queue<Event> events = new LinkedList<Event>();
+
+ public CollectorImpl()
+ {}
+
+ public Event peek()
+ {
+ return events.peek();
+ }
+
+ public void pop()
+ {
+ events.poll();
+ }
+
+ public EventImpl put(Event.Type type)
+ {
+ EventImpl event = new EventImpl(type);
+ events.add(event);
+ return event;
+ }
+
+}
Modified: qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/ConnectionImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/ConnectionImpl.java?rev=1574591&r1=1574590&r2=1574591&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/ConnectionImpl.java (original)
+++ qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/ConnectionImpl.java Wed Mar 5 17:55:11 2014
@@ -49,11 +49,11 @@ public class ConnectionImpl extends Endp
private DeliveryImpl _workHead;
private DeliveryImpl _workTail;
+ private TransportImpl _transport;
private DeliveryImpl _transportWorkHead;
private DeliveryImpl _transportWorkTail;
private String _localContainerId = "";
private String _localHostname = "";
- private boolean _bound;
private String _remoteContainer;
private String _remoteHostname;
private Symbol[] _offeredCapabilities;
@@ -64,6 +64,7 @@ public class ConnectionImpl extends Endp
private Map<Symbol, Object> _remoteProperties;
private Object _context;
+ private CollectorImpl _collector;
private static final Symbol[] EMPTY_SYMBOL_ARRAY = new Symbol[0];
@@ -191,7 +192,7 @@ public class ConnectionImpl extends Endp
_sessions = null;
}
- public void handleOpen(Open open)
+ void handleOpen(Open open)
{
// TODO - store state
setRemoteState(EndpointState.ACTIVE);
@@ -200,6 +201,10 @@ public class ConnectionImpl extends Endp
setRemoteDesiredCapabilities(open.getDesiredCapabilities());
setRemoteOfferedCapabilities(open.getOfferedCapabilities());
setRemoteProperties(open.getProperties());
+ EventImpl ev = put(Event.Type.CONNECTION_STATE);
+ if (ev != null) {
+ ev.init(this);
+ }
}
@@ -437,14 +442,14 @@ public class ConnectionImpl extends Endp
return new WorkSequence(_workHead);
}
- public void setBound(boolean bound)
+ void setTransport(TransportImpl transport)
{
- _bound = true;
+ _transport = transport;
}
- public boolean isBound()
+ TransportImpl getTransport()
{
- return _bound;
+ return _transport;
}
private class WorkSequence implements Iterator<DeliveryImpl>
@@ -516,6 +521,7 @@ public class ConnectionImpl extends Endp
void addTransportWork(DeliveryImpl delivery)
{
+ modified();
if (delivery._transportWork) return;
delivery.setTransportWorkNext(null);
@@ -562,4 +568,18 @@ public class ConnectionImpl extends Endp
_context = context;
}
+ public void collect(Collector collector)
+ {
+ _collector = (CollectorImpl) collector;
+ }
+
+ EventImpl put(Event.Type type)
+ {
+ if (_collector != null) {
+ return _collector.put(type);
+ } else {
+ return null;
+ }
+ }
+
}
Modified: qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/EndpointImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/EndpointImpl.java?rev=1574591&r1=1574590&r2=1574591&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/EndpointImpl.java (original)
+++ qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/EndpointImpl.java Wed Mar 5 17:55:11 2014
@@ -23,6 +23,7 @@ package org.apache.qpid.proton.engine.im
import org.apache.qpid.proton.amqp.transport.ErrorCondition;
import org.apache.qpid.proton.engine.EndpointState;
+import org.apache.qpid.proton.engine.Event;
import org.apache.qpid.proton.engine.ProtonJEndpoint;
public abstract class EndpointImpl implements ProtonJEndpoint
@@ -47,6 +48,7 @@ public abstract class EndpointImpl imple
case UNINITIALIZED:
_localState = EndpointState.ACTIVE;
}
+ modified();
}
public void close()
@@ -61,6 +63,7 @@ public abstract class EndpointImpl imple
case ACTIVE:
_localState = EndpointState.CLOSED;
}
+ modified();
}
public EndpointState getLocalState()
@@ -109,11 +112,24 @@ public abstract class EndpointImpl imple
void modified()
{
+ modified(true);
+ }
+
+ void modified(boolean emit)
+ {
if(!_modified)
{
_modified = true;
getConnectionImpl().addModified(this);
}
+
+ if (emit) {
+ ConnectionImpl conn = getConnectionImpl();
+ EventImpl ev = conn.put(Event.Type.TRANSPORT);
+ if (ev != null) {
+ ev.init(conn);
+ }
+ }
}
protected abstract ConnectionImpl getConnectionImpl();
Added: qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/EventImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/EventImpl.java?rev=1574591&view=auto
==============================================================================
--- qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/EventImpl.java (added)
+++ qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/EventImpl.java Wed Mar 5 17:55:11 2014
@@ -0,0 +1,109 @@
+/*
+ *
+ * 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.qpid.proton.engine.impl;
+
+import org.apache.qpid.proton.engine.Event;
+import org.apache.qpid.proton.engine.Connection;
+import org.apache.qpid.proton.engine.Session;
+import org.apache.qpid.proton.engine.Link;
+import org.apache.qpid.proton.engine.Delivery;
+import org.apache.qpid.proton.engine.Transport;
+
+/**
+ * EventImpl
+ *
+ */
+
+class EventImpl implements Event
+{
+
+ Type type;
+ Connection connection;
+ Session session;
+ Link link;
+ Delivery delivery;
+ Transport transport;
+
+ EventImpl(Type type)
+ {
+ this.type = type;
+ }
+
+ public Type getType()
+ {
+ return type;
+ }
+
+ public Connection getConnection()
+ {
+ return connection;
+ }
+
+ public Session getSession()
+ {
+ return session;
+ }
+
+ public Link getLink()
+ {
+ return link;
+ }
+
+ public Delivery getDelivery()
+ {
+ return delivery;
+ }
+
+ public Transport getTransport()
+ {
+ return transport;
+ }
+
+ void init(Transport transport)
+ {
+ this.transport = transport;
+ }
+
+ void init(Connection connection)
+ {
+ this.connection = connection;
+ init(((ConnectionImpl) connection).getTransport());
+ }
+
+ void init(Session session)
+ {
+ this.session = session;
+ init(session.getConnection());
+ }
+
+ void init(Link link)
+ {
+ this.link = link;
+ init(link.getSession());
+ }
+
+ void init(Delivery delivery)
+ {
+ this.delivery = delivery;
+ init(delivery.getLink());
+ }
+
+}
Modified: qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/LinkImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/LinkImpl.java?rev=1574591&r1=1574590&r2=1574591&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/LinkImpl.java (original)
+++ qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/LinkImpl.java Wed Mar 5 17:55:11 2014
@@ -63,18 +63,6 @@ public abstract class LinkImpl extends E
}
- public void open()
- {
- super.open();
- modified();
- }
-
- public void close()
- {
- super.close();
- modified();
- }
-
public String getName()
{
return _name;
@@ -203,7 +191,6 @@ public abstract class LinkImpl extends E
{
// TODO - should be an error if local state is ACTIVE
_source = source;
- modified();
}
public Target getTarget()
@@ -215,7 +202,6 @@ public abstract class LinkImpl extends E
{
// TODO - should be an error if local state is ACTIVE
_target = target;
- modified();
}
public Link next(EnumSet<EndpointState> local, EnumSet<EndpointState> remote)
Modified: qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/SessionImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/SessionImpl.java?rev=1574591&r1=1574590&r2=1574591&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/SessionImpl.java (original)
+++ qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/SessionImpl.java Wed Mar 5 17:55:11 2014
@@ -47,19 +47,6 @@ public class SessionImpl extends Endpoin
_node = _connection.addSessionEndpoint(this);
}
- public void open()
- {
- super.open();
- modified();
- }
-
- public void close()
- {
-
- super.close();
- modified();
- }
-
public SenderImpl sender(String name)
{
SenderImpl sender = _senders.get(name);
Modified: qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java?rev=1574591&r1=1574590&r2=1574591&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java (original)
+++ qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java Wed Mar 5 17:55:11 2014
@@ -48,6 +48,7 @@ import org.apache.qpid.proton.codec.Writ
import org.apache.qpid.proton.engine.Connection;
import org.apache.qpid.proton.engine.EndpointState;
import org.apache.qpid.proton.engine.EngineFactory;
+import org.apache.qpid.proton.engine.Event;
import org.apache.qpid.proton.engine.ProtonJTransport;
import org.apache.qpid.proton.engine.Sasl;
import org.apache.qpid.proton.engine.Ssl;
@@ -194,7 +195,7 @@ public class TransportImpl extends Endpo
public void bind(Connection conn)
{
// TODO - check if already bound
- ((ConnectionImpl) conn).setBound(true);
+ ((ConnectionImpl) conn).setTransport(this);
_connectionEndpoint = (ConnectionImpl) conn;
_localSessions = new TransportSession[_connectionEndpoint.getMaxChannels()+1];
@@ -472,7 +473,8 @@ public class TransportImpl extends Endpo
if(!delivery.isDone() &&
(delivery.getDataLength() > 0 || delivery != snd.current()) &&
- tpSession.hasOutgoingCredit() && tpLink.hasCredit())
+ tpSession.hasOutgoingCredit() && tpLink.hasCredit() &&
+ tpLink.getLocalHandle() != null)
{
UnsignedInteger deliveryId = tpSession.getOutgoingDeliveryId();
TransportDelivery tpDelivery = new TransportDelivery(deliveryId, delivery, tpLink);
@@ -976,7 +978,10 @@ public class TransportImpl extends Endpo
transportSession.setNextIncomingId(begin.getNextOutgoingId());
_remoteSessions[channel] = transportSession;
-
+ EventImpl ev = _connectionEndpoint.put(Event.Type.SESSION_STATE);
+ if (ev != null) {
+ ev.init(session);
+ }
}
}
@@ -1031,6 +1036,11 @@ public class TransportImpl extends Endpo
transportSession.addLinkRemoteHandle(transportLink, attach.getHandle());
}
+
+ EventImpl ev = _connectionEndpoint.put(Event.Type.LINK_STATE);
+ if (ev != null) {
+ ev.init(link);
+ }
}
}
@@ -1100,6 +1110,11 @@ public class TransportImpl extends Endpo
{
link.getRemoteCondition().copyFrom(detach.getError());
}
+
+ EventImpl ev = _connectionEndpoint.put(Event.Type.LINK_STATE);
+ if (ev != null) {
+ ev.init(link);
+ }
}
else
{
@@ -1127,6 +1142,11 @@ public class TransportImpl extends Endpo
{
session.getRemoteCondition().copyFrom(errorCondition);
}
+
+ EventImpl ev = _connectionEndpoint.put(Event.Type.SESSION_STATE);
+ if (ev != null) {
+ ev.init(session);
+ }
}
}
@@ -1142,6 +1162,11 @@ public class TransportImpl extends Endpo
{
_connectionEndpoint.getRemoteCondition().copyFrom(close.getError());
}
+
+ EventImpl ev = _connectionEndpoint.put(Event.Type.CONNECTION_STATE);
+ if (ev != null) {
+ ev.init(_connectionEndpoint);
+ }
}
}
Modified: qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportLink.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportLink.java?rev=1574591&r1=1574590&r2=1574591&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportLink.java (original)
+++ qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportLink.java Wed Mar 5 17:55:11 2014
@@ -23,6 +23,7 @@ package org.apache.qpid.proton.engine.im
import org.apache.qpid.proton.amqp.UnsignedInteger;
import org.apache.qpid.proton.amqp.transport.Flow;
+import org.apache.qpid.proton.engine.Event;
class TransportLink<T extends LinkImpl>
{
@@ -121,6 +122,10 @@ class TransportLink<T extends LinkImpl>
_remoteLinkCredit = flow.getLinkCredit();
+ EventImpl ev = _link.getConnectionImpl().put(Event.Type.LINK_FLOW);
+ if (ev != null) {
+ ev.init(_link);
+ }
}
void setLinkCredit(UnsignedInteger linkCredit)
Modified: qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportSession.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportSession.java?rev=1574591&r1=1574590&r2=1574591&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportSession.java (original)
+++ qpid/proton/trunk/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportSession.java Wed Mar 5 17:55:11 2014
@@ -29,6 +29,7 @@ import org.apache.qpid.proton.amqp.trans
import org.apache.qpid.proton.amqp.transport.Flow;
import org.apache.qpid.proton.amqp.transport.Role;
import org.apache.qpid.proton.amqp.transport.Transfer;
+import org.apache.qpid.proton.engine.Event;
class TransportSession
{
@@ -304,7 +305,12 @@ class TransportSession
// this will cause a flow to happen
if (_incomingWindowSize.equals(UnsignedInteger.ZERO)) {
- delivery.getLink().modified();
+ delivery.getLink().modified(false);
+ }
+
+ EventImpl ev = getSession().getConnection().put(Event.Type.DELIVERY);
+ if (ev != null) {
+ ev.init(delivery);
}
}
@@ -387,6 +393,11 @@ class TransportSession
unsettledDeliveries.remove(id);
}
delivery.updateWork();
+
+ EventImpl ev = getSession().getConnection().put(Event.Type.DELIVERY);
+ if (ev != null) {
+ ev.init(delivery);
+ }
}
id = id.add(UnsignedInteger.ONE);
}
@@ -415,12 +426,12 @@ class TransportSession
if(transportDelivery.getTransportLink().getLink() instanceof ReceiverImpl)
{
_unsettledIncomingDeliveriesById.remove(transportDelivery.getDeliveryId());
- getSession().modified();
+ getSession().modified(false);
}
else
{
_unsettledOutgoingDeliveriesById.remove(transportDelivery.getDeliveryId());
- getSession().modified();
+ getSession().modified(false);
}
}
Modified: qpid/proton/trunk/proton-j/src/main/resources/cengine.py
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/main/resources/cengine.py?rev=1574591&r1=1574590&r2=1574591&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/src/main/resources/cengine.py (original)
+++ qpid/proton/trunk/proton-j/src/main/resources/cengine.py Wed Mar 5 17:55:11 2014
@@ -928,12 +928,44 @@ def pn_transport_close_tail(trans):
def pn_transport_error(trans):
return trans.error
-PN_CONNECTION_STATE = "PN_CONNECTION_STATE"
-PN_SESSION_STATE = "PN_SESSION_STATE"
-PN_LINK_STATE = "PN_LINK_STATE"
-PN_LINK_FLOW = "PN_LINK_FLOW"
-PN_DELIVERY = "PN_DELIVERY"
-PN_TRANSPORT = "PN_TRANSPORT"
+from org.apache.qpid.proton.engine import Event
+
+PN_CONNECTION_STATE = Event.Type.CONNECTION_STATE
+PN_SESSION_STATE = Event.Type.SESSION_STATE
+PN_LINK_STATE = Event.Type.LINK_STATE
+PN_LINK_FLOW = Event.Type.LINK_FLOW
+PN_DELIVERY = Event.Type.DELIVERY
+PN_TRANSPORT = Event.Type.TRANSPORT
def pn_collector():
- raise Skipped()
+ return Proton.collector()
+
+def pn_connection_collect(conn, coll):
+ conn.impl.collect(coll)
+
+def pn_collector_peek(coll):
+ return coll.peek()
+
+def pn_collector_pop(coll):
+ coll.pop()
+
+def pn_event_connection(event):
+ return wrap(event.getConnection(), pn_connection_wrapper)
+
+def pn_event_session(event):
+ return wrap(event.getSession(), pn_session_wrapper)
+
+def pn_event_link(event):
+ return wrap(event.getLink(), pn_link_wrapper)
+
+def pn_event_delivery(event):
+ return wrap(event.getDelivery(), pn_delivery_wrapper)
+
+def pn_event_transport(event):
+ return wrap(event.getTransport(), pn_transport_wrapper)
+
+def pn_event_type(event):
+ return event.getType()
+
+def pn_event_type_name(etype):
+ return str(etype)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org