From c32eeb072756c62db34ec83f643a486f81b039f8 Mon Sep 17 00:00:00 2001 From: markpollack Date: Sun, 3 May 2009 04:55:39 +0000 Subject: [PATCH] Add interfaces for ConnectionFactory, Connection, Session, MessageConsumer, MessageProducer, and TopicSubscriber and EMS implementations Add caching infrastructure --- .../Messaging/Ems/Common/EmsConnection.cs | 131 +++ .../Ems/Common/EmsConnectionFactory.cs | 217 +++++ .../Ems/Common/EmsMessageConsumer.cs | 78 ++ .../Ems/Common/EmsMessageProducer.cs | 119 +++ .../Messaging/Ems/Common/EmsSession.cs | 216 +++++ .../Ems/Common/EmsTopicSubscriber.cs | 46 ++ .../Messaging/Ems/Common/IConnection.cs | 52 ++ .../Ems/Common/IConnectionFactory.cs | 65 ++ .../Messaging/Ems/Common/IMessageConsumer.cs | 37 + .../Messaging/Ems/Common/IMessageProducer.cs | 52 ++ .../Messaging/Ems/Common/ISession.cs | 77 ++ .../Messaging/Ems/Common/ITopicSubscriber.cs | 30 + ...ListenerContainerObjectDefinitionParser.cs | 11 +- .../Ems/Connections/CachedMessageConsumer .cs | 136 +++ .../Ems/Connections/CachedMessageProducer.cs | 237 ++++++ .../Ems/Connections/CachedSession.cs | 678 +++++++++++++++ .../Connections/CachingConnectionFactory.cs | 278 +++++++ .../Ems/Connections/ConnectionFactoryUtils.cs | 83 +- .../Ems/Connections/EmsResourceHolder.cs | 43 +- .../Ems/Connections/EmsTransactionManager.cs | 19 +- .../Ems/Connections/IDecoratorSession.cs | 42 + .../Connections/ISmartConnectionFactory.cs | 43 + .../Connections/SingleConnectionFactory.cs | 777 ++++++++++++++++++ .../Messaging/Ems/Core/BrowserDelegate.cs | 3 +- .../Messaging/Ems/Core/EmsGatewaySupport.cs | 5 +- .../Messaging/Ems/Core/EmsTemplate.cs | 124 +-- .../Messaging/Ems/Core/IBrowserCallback.cs | 3 +- .../Messaging/Ems/Core/IEmsOperations.cs | 1 - .../Messaging/Ems/Core/IMessageCreator.cs | 3 +- .../Messaging/Ems/Core/IProducerCallback.cs | 4 +- .../Messaging/Ems/Core/ISessionCallback.cs | 4 +- .../Ems/Core/MessageCreatorDelegate.cs | 3 +- .../Messaging/Ems/Core/ProducerDelegate.cs | 4 +- .../Messaging/Ems/Core/SessionDelegate.cs | 4 +- .../Ems/Listener/AbstractListenerContainer.cs | 11 +- .../AbstractMessageListenerContainer.cs | 23 +- .../Adapter/MessageListenerAdapter.cs | 17 +- .../Listener/ISessionAwareMessageListener.cs | 3 +- .../LocallyExposedEmsResourceHolder.cs | 4 +- .../SimpleMessageListenerContainer.cs | 23 +- .../Support/Converter/IMessageConverter.cs | 3 +- .../Converter/SimpleMessageConverter.cs | 11 +- .../Support/Converter/XmlMessageConverter.cs | 3 +- .../DynamicDestinationResolver.cs | 12 +- .../Destinations/EmsDestinationAccessor.cs | 3 +- .../Destinations/IDestinationResolver.cs | 3 +- .../Messaging/Ems/Support/EmsAccessor.cs | 11 +- .../Messaging/Ems/Support/EmsUtils.cs | 45 +- .../SimpleMessageListenerContainerTests.cs | 6 +- .../SimpleMessageListenerContainerTests.xml | 5 +- .../Spring.Messaging.Ems.Tests.2005.csproj | 4 + 51 files changed, 3612 insertions(+), 200 deletions(-) create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsConnection.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsConnectionFactory.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsMessageConsumer.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsMessageProducer.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsSession.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsTopicSubscriber.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/IConnection.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/IConnectionFactory.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/IMessageConsumer.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/IMessageProducer.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/ISession.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/ITopicSubscriber.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/CachedMessageConsumer .cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/CachedMessageProducer.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/CachedSession.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/CachingConnectionFactory.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/IDecoratorSession.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ISmartConnectionFactory.cs create mode 100644 src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/SingleConnectionFactory.cs diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsConnection.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsConnection.cs new file mode 100644 index 00000000..491c6140 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsConnection.cs @@ -0,0 +1,131 @@ +#region License + +/* + * Copyright © 2002-2009 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using Common.Logging; +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Common +{ + public class EmsConnection : IConnection + { + #region Logging + + private readonly ILog logger = LogManager.GetLogger(typeof(EmsConnection)); + + #endregion + + private Connection nativeConnection; + + public EmsConnection(Connection connection) + { + this.nativeConnection = connection; + this.nativeConnection.ExceptionHandler += HandleEmsException; + } + + + + #region Implementation of IConnection + + public Connection NativeConnection + { + get { return this.nativeConnection; } + } + + public event EMSExceptionHandler EMSExceptionHandler; + + public string ActiveURL + { + get { return nativeConnection.ActiveURL; } + } + + public string ClientID + { + get { return nativeConnection.ClientID; } + set { nativeConnection.ClientID = value; } + } + + public long ConnID + { + get { return nativeConnection.ConnID; } + } + + public IExceptionListener ExceptionListener + { + get { return nativeConnection.ExceptionListener; } + set { nativeConnection.ExceptionListener = value; } + } + + public bool IsClosed + { + get { return nativeConnection.IsClosed; } + } + + public bool IsSecure + { + get { return nativeConnection.IsSecure; } + } + + public ConnectionMetaData MetaData + { + get { return nativeConnection.MetaData; } + } + + public void Close() + { + nativeConnection.Close(); + } + + public ISession CreateSession(bool transacted, int acknowledgeMode) + { + Session nativeSession = nativeConnection.CreateSession(transacted, acknowledgeMode); + return new EmsSession(nativeSession); + } + + public ISession CreateSession(bool transacted, SessionMode acknowledgeMode) + { + Session nativeSession = nativeConnection.CreateSession(transacted, acknowledgeMode); + return new EmsSession(nativeSession); + } + + public void Start() + { + nativeConnection.Start(); + } + + public void Stop() + { + nativeConnection.Stop(); + } + + #endregion + + private void HandleEmsException(object sender, EMSExceptionEventArgs arg) + { + if (EMSExceptionHandler != null) + { + EMSExceptionHandler(sender, arg); + } + else + { + logger.Error("No exception handler registered with EmsConnection wrapper class.", arg.Exception); + } + } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsConnectionFactory.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsConnectionFactory.cs new file mode 100644 index 00000000..a9bf3f79 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsConnectionFactory.cs @@ -0,0 +1,217 @@ +#region License + +/* + * Copyright © 2002-2009 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using System.IO; +using System.Runtime.Serialization; +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Common +{ + public class EmsConnectionFactory : IConnectionFactory + { + private ConnectionFactory nativeConnectionFactory; + + public EmsConnectionFactory(ConnectionFactory nativeConnectionFactory) + { + this.nativeConnectionFactory = nativeConnectionFactory; + } + + + #region Implementation of ISerializable + + public void GetObjectData(SerializationInfo info, StreamingContext context) + { + nativeConnectionFactory.GetObjectData(info, context); + } + + #endregion + + #region Implementation of ICloneable + + public object Clone() + { + return nativeConnectionFactory.Clone(); + } + + #endregion + + #region Implementation of IConnectionFactory + + public ConnectionFactory NativeConnectionFactory + { + get { return this.nativeConnectionFactory; } + } + + public IConnection CreateConnection() + { + Connection nativeConnection = nativeConnectionFactory.CreateConnection(); + return new EmsConnection(nativeConnection); + } + + public IConnection CreateConnection(string userName, string password) + { + Connection nativeConnection = nativeConnectionFactory.CreateConnection(userName, password); + return new EmsConnection(nativeConnection); + } + + public object GetCertificateStore() + { + return nativeConnectionFactory.GetCertificateStore(); + } + + public string GetSSLProxyHost() + { + return nativeConnectionFactory.GetSSLProxyHost(); + } + + public string GetSSLProxyPassword() + { + return nativeConnectionFactory.GetSSLProxyPassword(); + } + + public int GetSSLProxyPort() + { + return nativeConnectionFactory.GetSSLProxyPort(); + } + + public string GetSSLProxyUser() + { + return nativeConnectionFactory.GetSSLProxyUser(); + } + + public void SetCertificateStoreType(EMSSSLStoreType type, object storeInfo) + { + nativeConnectionFactory.SetCertificateStoreType(type, storeInfo); + } + + public void SetClientID(string clientID) + { + nativeConnectionFactory.SetClientID(clientID); + } + + public void SetClientTracer(StreamWriter tracer) + { + nativeConnectionFactory.SetClientTracer(tracer); + } + + public void SetConnAttemptCount(int attempts) + { + nativeConnectionFactory.SetConnAttemptCount(attempts); + } + + public void SetConnAttemptDelay(int delay) + { + nativeConnectionFactory.SetConnAttemptDelay(delay); + } + + public void SetConnAttemptTimeout(int timeout) + { + nativeConnectionFactory.SetConnAttemptTimeout(timeout); + } + + public void SetHostNameVerifier(EMSSSLHostNameVerifier verifier) + { + nativeConnectionFactory.SetHostNameVerifier(verifier); + } + + public void SetMetric(int metric) + { + nativeConnectionFactory.SetMetric(metric); + } + + public void SetMulticastDaemon(string port) + { + nativeConnectionFactory.SetMulticastDaemon(port); + } + + public void SetMulticastEnabled(bool enabled) + { + nativeConnectionFactory.SetMulticastEnabled(enabled); + } + + public void SetReconnAttemptCount(int attempts) + { + nativeConnectionFactory.SetReconnAttemptCount(attempts); + } + + public void SetReconnAttemptDelay(int delay) + { + nativeConnectionFactory.SetReconnAttemptDelay(delay); + } + + public void SetReconnAttemptTimeout(int timeout) + { + nativeConnectionFactory.SetReconnAttemptTimeout(timeout); + } + + public void SetServerUrl(string serverUrl) + { + nativeConnectionFactory.SetServerUrl(serverUrl); + } + + public void SetSSLAuthOnly(bool authOnly) + { + nativeConnectionFactory.SetSSLAuthOnly(authOnly); + } + + public void SetSSLProxy(string host, int port) + { + nativeConnectionFactory.SetSSLProxy(host, port); + } + + public void SetSSLProxyAuth(string username, string password) + { + nativeConnectionFactory.SetSSLProxyAuth(username, password); + } + + public void SetSSLTrace(bool trace) + { + nativeConnectionFactory.SetSSLTrace(trace); + } + + public void SetTargetHostName(string targetHostName) + { + nativeConnectionFactory.SetTargetHostName(targetHostName); + } + + public void SetUserName(string username) + { + nativeConnectionFactory.SetUserName(username); + } + + public void SetUserPassword(string password) + { + nativeConnectionFactory.SetUserPassword(password); + } + + public FactoryLoadBalanceMetric Metric + { + get + { + return nativeConnectionFactory.Metric; + } + set { + nativeConnectionFactory.Metric = value; + } + } + + #endregion + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsMessageConsumer.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsMessageConsumer.cs new file mode 100644 index 00000000..2bedae6f --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsMessageConsumer.cs @@ -0,0 +1,78 @@ +#region License + +/* + * Copyright © 2002-2009 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using System; +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Common +{ + public class EmsMessageConsumer : IMessageConsumer + { + protected readonly MessageConsumer nativeMessageConsumer; + + public EmsMessageConsumer(MessageConsumer messageConsumer) + { + nativeMessageConsumer = messageConsumer; + nativeMessageConsumer.MessageHandler += MessageHandler; + } + + #region Implementation of IMessageConsumer + + public MessageConsumer NativeMessageConsumer + { + get { return this.nativeMessageConsumer; } + } + + public event EMSMessageHandler MessageHandler; + + public IMessageListener MessageListener + { + get { return nativeMessageConsumer.MessageListener; } + set { nativeMessageConsumer.MessageListener = value; } + } + + public string MessageSelector + { + get { return nativeMessageConsumer.MessageSelector; } + } + + public void Close() + { + nativeMessageConsumer.Close(); + } + + public Message Receive() + { + return nativeMessageConsumer.Receive(); + } + + public Message Receive(long timeout) + { + return nativeMessageConsumer.Receive(timeout); + } + + public Message ReceiveNoWait() + { + return nativeMessageConsumer.ReceiveNoWait(); + } + + #endregion + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsMessageProducer.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsMessageProducer.cs new file mode 100644 index 00000000..e854db1b --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsMessageProducer.cs @@ -0,0 +1,119 @@ +#region License + +/* + * Copyright © 2002-2009 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Common +{ + public class EmsMessageProducer : IMessageProducer + { + private MessageProducer nativeMessageProducer; + + public EmsMessageProducer(MessageProducer messageProducer) + { + this.nativeMessageProducer = messageProducer; + } + + #region Implementation of IMessageProducer + + public MessageProducer NativeMessageProducer + { + get { return this.nativeMessageProducer; } + } + + public void Close() + { + nativeMessageProducer.Close(); + } + + public void Send(Message message) + { + nativeMessageProducer.Send(message); + } + + public void Send(Destination dest, Message message) + { + nativeMessageProducer.Send(dest, message); + } + + public void Send(Message message, int deliveryMode, int priority, long timeToLive) + { + nativeMessageProducer.Send(message, deliveryMode, priority, timeToLive); + } + + public void Send(Message message, MessageDeliveryMode deliveryMode, int priority, long timeToLive) + { + nativeMessageProducer.Send(message, deliveryMode, priority, timeToLive); + } + + public void Send(Destination dest, Message message, int deliveryMode, int priority, long timeToLive) + { + nativeMessageProducer.Send(dest, message, deliveryMode, priority, timeToLive); + } + + public void Send(Destination dest, Message message, MessageDeliveryMode deliveryMode, int priority, long timeToLive) + { + nativeMessageProducer.Send(dest, message, deliveryMode, priority, timeToLive); + } + + public int DeliveryMode + { + get { return nativeMessageProducer.DeliveryMode; } + set { nativeMessageProducer.DeliveryMode = value; } + } + + public Destination Destination + { + get { return nativeMessageProducer.Destination; } + } + + public bool DisableMessageID + { + get { return nativeMessageProducer.DisableMessageID; } + set { nativeMessageProducer.DisableMessageID = value; } + } + + public bool DisableMessageTimestamp + { + get { return nativeMessageProducer.DisableMessageTimestamp; } + set { nativeMessageProducer.DisableMessageTimestamp = value; } + } + + public MessageDeliveryMode MsgDeliveryMode + { + get { return nativeMessageProducer.MsgDeliveryMode; } + set { nativeMessageProducer.MsgDeliveryMode = value; } + } + + public int Priority + { + get { return nativeMessageProducer.Priority; } + set { nativeMessageProducer.Priority = value; } + } + + public long TimeToLive + { + get { return nativeMessageProducer.TimeToLive; } + set { nativeMessageProducer.TimeToLive = value; } + } + + #endregion + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsSession.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsSession.cs new file mode 100644 index 00000000..357bd344 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsSession.cs @@ -0,0 +1,216 @@ +#region License + +/* + * Copyright © 2002-2009 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Common +{ + public class EmsSession : ISession + { + private Session nativeSession; + + public EmsSession(Session session) + { + this.nativeSession = session; + } + + #region Implementation of ISession + + public Session NativeSession + { + get { return this.nativeSession; } + } + + public void Close() + { + nativeSession.Close(); + } + + public void Commit() + { + nativeSession.Commit(); + } + + public QueueBrowser CreateBrowser(Queue queue) + { + return nativeSession.CreateBrowser(queue); + } + + public QueueBrowser CreateBrowser(Queue queue, string messageSelector) + { + return nativeSession.CreateBrowser(queue, messageSelector); + } + + public BytesMessage CreateBytesMessage() + { + return nativeSession.CreateBytesMessage(); + } + + public IMessageConsumer CreateConsumer(Destination dest) + { + return new EmsMessageConsumer(nativeSession.CreateConsumer(dest)); + + } + + public IMessageConsumer CreateConsumer(Destination dest, string messageSelector) + { + return new EmsMessageConsumer(nativeSession.CreateConsumer(dest, messageSelector)); + } + + public IMessageConsumer CreateConsumer(Destination dest, string messageSelector, bool noLocal) + { + return new EmsMessageConsumer(nativeSession.CreateConsumer(dest, messageSelector, noLocal)); + } + + public ITopicSubscriber CreateDurableSubscriber(Topic topic, string name) + { + return new EmsTopicSubscriber(nativeSession.CreateDurableSubscriber(topic, name)); + } + + public ITopicSubscriber CreateDurableSubscriber(Topic topic, string name, string messageSelector, bool noLocal) + { + return new EmsTopicSubscriber(nativeSession.CreateDurableSubscriber(topic, name, messageSelector, noLocal)); + } + + public MapMessage CreateMapMessage() + { + return nativeSession.CreateMapMessage(); + } + + public Message CreateMessage() + { + return nativeSession.CreateMessage(); + } + + public ObjectMessage CreateObjectMessage() + { + return nativeSession.CreateObjectMessage(); + } + + public ObjectMessage CreateObjectMessage(object obj) + { + return nativeSession.CreateObjectMessage(obj); + } + + public IMessageProducer CreateProducer(Destination dest) + { + return new EmsMessageProducer(nativeSession.CreateProducer(dest)); + } + + public Queue CreateQueue(string queueName) + { + return nativeSession.CreateQueue(queueName); + } + + public StreamMessage CreateStreamMessage() + { + return nativeSession.CreateStreamMessage(); + } + + public TemporaryQueue CreateTemporaryQueue() + { + return nativeSession.CreateTemporaryQueue(); + } + + public TemporaryTopic CreateTemporaryTopic() + { + return nativeSession.CreateTemporaryTopic(); + } + + public TextMessage CreateTextMessage() + { + return nativeSession.CreateTextMessage(); + } + + public TextMessage CreateTextMessage(string text) + { + return nativeSession.CreateTextMessage(text); + } + + public Topic CreateTopic(string topicName) + { + return nativeSession.CreateTopic(topicName); + } + + public void Recover() + { + nativeSession.Recover(); + } + + public void Rollback() + { + nativeSession.Rollback(); + } + + public void Run() + { + nativeSession.Run(); + } + + public void Unsubscribe(string name) + { + nativeSession.Unsubscribe(name); + } + + public int AcknowledgeMode + { + get { return nativeSession.AcknowledgeMode; } + } + + // TODO + public Connection Connection + { + get { return nativeSession.Connection; } + } + + public bool IsClosed + { + get { return nativeSession.IsClosed; } + } + + public bool IsTransacted + { + get { return nativeSession.IsTransacted; } + } + + public IMessageListener MessageListener + { + get { return nativeSession.MessageListener; } + set { nativeSession.MessageListener = value; } + } + + public long SessID + { + get { return nativeSession.SessID; } + } + + public SessionMode SessionAcknowledgeMode + { + get { return nativeSession.SessionAcknowledgeMode; } + } + + public bool Transacted + { + get { return nativeSession.Transacted; } + } + + #endregion + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsTopicSubscriber.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsTopicSubscriber.cs new file mode 100644 index 00000000..c9bd03bf --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/EmsTopicSubscriber.cs @@ -0,0 +1,46 @@ +#region License + +/* + * Copyright © 2002-2009 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Common +{ + public class EmsTopicSubscriber : EmsMessageConsumer, ITopicSubscriber + { + public EmsTopicSubscriber(TopicSubscriber topicSubscriber) + : base(topicSubscriber) + { + } + + #region Implementation of ITopicSubscriber + + public bool NoLocal + { + get { return ((TopicSubscriber) nativeMessageConsumer).NoLocal; } + } + + public Topic Topic + { + get { return ((TopicSubscriber) nativeMessageConsumer).Topic; } + } + + #endregion + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/IConnection.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/IConnection.cs new file mode 100644 index 00000000..ab9d6e36 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/IConnection.cs @@ -0,0 +1,52 @@ +#region License + +/* + * Copyright © 2002-2009 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using System; +using System.ComponentModel; +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Common +{ + public interface IConnection + { + Connection NativeConnection { get; } + event EMSExceptionHandler EMSExceptionHandler; + string ActiveURL { get; } + string ClientID { get; set; } + + [EditorBrowsable(EditorBrowsableState.Never)] + long ConnID { get; } + + IExceptionListener ExceptionListener { get; set; } + bool IsClosed { get; } + + [EditorBrowsable(EditorBrowsableState.Never), Obsolete("EMS clients should no longer use this method; it has been deprecated.")] + bool IsSecure { get; } + + ConnectionMetaData MetaData { get; } + + void Close(); + ISession CreateSession(bool transacted, int acknowledgeMode); + ISession CreateSession(bool transacted, SessionMode acknowledgeMode); + void Start(); + void Stop(); + string ToString(); + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/IConnectionFactory.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/IConnectionFactory.cs new file mode 100644 index 00000000..3952ec85 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/IConnectionFactory.cs @@ -0,0 +1,65 @@ +#region License + +/* + * Copyright © 2002-2009 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using System; +using System.IO; +using System.Runtime.Serialization; +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Common +{ + public interface IConnectionFactory : ISerializable, ICloneable + { + ConnectionFactory NativeConnectionFactory { get; } + + IConnection CreateConnection(); + IConnection CreateConnection(string userName, string password); + object GetCertificateStore(); + string GetSSLProxyHost(); + string GetSSLProxyPassword(); + int GetSSLProxyPort(); + string GetSSLProxyUser(); + + void SetCertificateStoreType(EMSSSLStoreType type, object storeInfo); + void SetClientID(string clientID); + void SetClientTracer(StreamWriter tracer); + void SetConnAttemptCount(int attempts); + void SetConnAttemptDelay(int delay); + void SetConnAttemptTimeout(int timeout); + void SetHostNameVerifier(EMSSSLHostNameVerifier verifier); + void SetMetric(int metric); + void SetMulticastDaemon(string port); + void SetMulticastEnabled(bool enabled); + void SetReconnAttemptCount(int attempts); + void SetReconnAttemptDelay(int delay); + void SetReconnAttemptTimeout(int timeout); + void SetServerUrl(string serverUrl); + void SetSSLAuthOnly(bool authOnly); + void SetSSLProxy(string host, int port); + void SetSSLProxyAuth(string username, string password); + void SetSSLTrace(bool trace); + void SetTargetHostName(string targetHostName); + void SetUserName(string username); + void SetUserPassword(string password); + + string ToString(); + FactoryLoadBalanceMetric Metric { get; set; } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/IMessageConsumer.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/IMessageConsumer.cs new file mode 100644 index 00000000..34f55088 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/IMessageConsumer.cs @@ -0,0 +1,37 @@ +#region License + +/* + * Copyright © 2002-2009 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Common +{ + public interface IMessageConsumer + { + MessageConsumer NativeMessageConsumer { get; } + event EMSMessageHandler MessageHandler; + IMessageListener MessageListener { get; set; } + string MessageSelector { get; } + void Close(); + Message Receive(); + Message Receive(long timeout); + Message ReceiveNoWait(); + string ToString(); + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/IMessageProducer.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/IMessageProducer.cs new file mode 100644 index 00000000..78ba7a0e --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/IMessageProducer.cs @@ -0,0 +1,52 @@ +#region License + +/* + * Copyright © 2002-2009 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Common +{ + public interface IMessageProducer + { + MessageProducer NativeMessageProducer { get; } + + void Close(); + void Send(Message message); + void Send(TIBCO.EMS.Destination dest, Message message); + void Send(Message message, int deliveryMode, int priority, long timeToLive); + void Send(Message message, MessageDeliveryMode deliveryMode, int priority, long timeToLive); + void Send(TIBCO.EMS.Destination dest, Message message, int deliveryMode, int priority, long timeToLive); + void Send(TIBCO.EMS.Destination dest, Message message, MessageDeliveryMode deliveryMode, int priority, long timeToLive); + string ToString(); + int DeliveryMode { get; set; } + TIBCO.EMS.Destination Destination { get; } + bool DisableMessageID { get; set; } + bool DisableMessageTimestamp { get; set; } + MessageDeliveryMode MsgDeliveryMode { get; set; } + int Priority { get; set; } + + /// + /// Gets or sets the the default length of time in milliseconds from its dispatch time + /// that a produced message should be retained by the message system. + /// + /// Time to live is set to zero by default. + /// The message time to live in milliseconds; zero is unlimited + long TimeToLive { get; set; } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/ISession.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/ISession.cs new file mode 100644 index 00000000..00359c90 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/ISession.cs @@ -0,0 +1,77 @@ +#region License + +/* + * Copyright © 2002-2009 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using System; +using System.ComponentModel; +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Common +{ + public interface ISession + { + Session NativeSession { get; } + void Close(); + void Commit(); + QueueBrowser CreateBrowser(Queue queue); + QueueBrowser CreateBrowser(Queue queue, string messageSelector); + + IMessageConsumer CreateConsumer(Destination dest); + IMessageConsumer CreateConsumer(Destination dest, string messageSelector); + IMessageConsumer CreateConsumer(Destination dest, string messageSelector, bool noLocal); + ITopicSubscriber CreateDurableSubscriber(Topic topic, string name); + ITopicSubscriber CreateDurableSubscriber(Topic topic, string name, string messageSelector, bool noLocal); + IMessageProducer CreateProducer(Destination dest); + + + Queue CreateQueue(string queueName); + Topic CreateTopic(string topicName); + TemporaryQueue CreateTemporaryQueue(); + TemporaryTopic CreateTemporaryTopic(); + + Message CreateMessage(); + TextMessage CreateTextMessage(); + TextMessage CreateTextMessage(string text); + MapMessage CreateMapMessage(); + BytesMessage CreateBytesMessage(); + ObjectMessage CreateObjectMessage(); + ObjectMessage CreateObjectMessage(object obj); + StreamMessage CreateStreamMessage(); + + + void Recover(); + void Rollback(); + + [EditorBrowsable(EditorBrowsableState.Never), Obsolete("Ordinary JMS clients should not use this method.")] + void Run(); + + void Unsubscribe(string name); + int AcknowledgeMode { get; } + TIBCO.EMS.Connection Connection { get; } + bool IsClosed { get; } + bool IsTransacted { get; } + + [Obsolete("Use MessageConsumer.MessageListener instead.")] + IMessageListener MessageListener { get; set; } + + long SessID { get; } + SessionMode SessionAcknowledgeMode { get; } + bool Transacted { get; } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/ITopicSubscriber.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/ITopicSubscriber.cs new file mode 100644 index 00000000..f7787faf --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Common/ITopicSubscriber.cs @@ -0,0 +1,30 @@ +#region License + +/* + * Copyright © 2002-2009 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Common +{ + public interface ITopicSubscriber : IMessageConsumer + { + bool NoLocal { get; } + Topic Topic { get; } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Config/MessageListenerContainerObjectDefinitionParser.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Config/MessageListenerContainerObjectDefinitionParser.cs index 25d76cc2..2f97d31a 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Config/MessageListenerContainerObjectDefinitionParser.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Config/MessageListenerContainerObjectDefinitionParser.cs @@ -21,8 +21,7 @@ using System; using System.Xml; using Spring.Core.TypeResolution; -using TIBCO.EMS; -using Spring.Core.TypeConversion; +//using TIBCO.EMS; using Spring.Messaging.Ems.Listener; using Spring.Messaging.Ems.Listener.Adapter; using Spring.Objects.Factory.Config; @@ -373,15 +372,15 @@ namespace Spring.Messaging.Ems.Config string acknowledge = element.GetAttribute(ACKNOWLEDGE_ATTRIBUTE); if (acknowledge.Equals(ACKNOWLEDGE_TRANSACTED)) { - return Session.SESSION_TRANSACTED; + return TIBCO.EMS.Session.SESSION_TRANSACTED; } else if (acknowledge.Equals(ACKNOWLEDGE_DUPS_OK)) { - return Session.DUPS_OK_ACKNOWLEDGE; + return TIBCO.EMS.Session.DUPS_OK_ACKNOWLEDGE; } else if (acknowledge.Equals(ACKNOWLEDGE_CLIENT)) { - return Session.CLIENT_ACKNOWLEDGE; + return TIBCO.EMS.Session.CLIENT_ACKNOWLEDGE; } //TODO other ack modes. else if (!acknowledge.Equals(ACKNOWLEDGE_AUTO)) @@ -391,7 +390,7 @@ namespace Spring.Messaging.Ems.Config acknowledge + "]: only \"auto\", \"client\", \"dups-ok\" and \"transacted\" supported."); } - return Session.AUTO_ACKNOWLEDGE; + return TIBCO.EMS.Session.AUTO_ACKNOWLEDGE; } private string ParseConcurrency(XmlElement ele, ParserContext parserContext) diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/CachedMessageConsumer .cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/CachedMessageConsumer .cs new file mode 100644 index 00000000..0aca247b --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/CachedMessageConsumer .cs @@ -0,0 +1,136 @@ +#region License + +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using System; +using Spring.Messaging.Ems.Common; +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Connections +{ + /// + /// NMS MessageConsumer decorator that adapts all calls + /// to a shared MessageConsumer instance underneath. + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public class CachedMessageConsumer : IMessageConsumer + { + private IMessageConsumer target; + + /// + /// Initializes a new instance of the class. + /// + /// The target. + public CachedMessageConsumer(IMessageConsumer target) + { + this.target = target; + } + + + /// + /// Gets the target MessageConsumer, the consumer we are 'wrapping' + /// + /// The target MessageConsumer. + public IMessageConsumer Target + { + get { return target; } + } + + /// + /// Occurs when a message is received. + /// + public event EMSMessageHandler MessageHandler + { + add + { + target.MessageHandler += value; + } + remove + { + target.MessageHandler -= value; + } + } + + + public MessageConsumer NativeMessageConsumer + { + get { return target.NativeMessageConsumer; } + } + + public IMessageListener MessageListener + { + get { return target.MessageListener; } + set { target.MessageListener = value; } + } + + public string MessageSelector + { + get { return target.MessageSelector; } + } + + /// + /// Receives the next message produced for this message consumer. + /// + /// the next message produced for this message consumer, , or null if this message consumer is concurrently closed + public Message Receive() + { + return this.target.Receive(); + } + + /// + /// Receives the next message that arrives within the specified timeout interval. + /// + /// The timeout value. + /// the next message produced for this message consumer, or null if the timeout expires or this message consumer is concurrently closed + public Message Receive(long timeout) + { + return this.target.Receive(timeout); + } + + /// + /// Receives the next message if one is immediately available. + /// + /// the next message produced for this message consumer, or null if one is not available + public Message ReceiveNoWait() + { + return this.target.ReceiveNoWait(); + } + + /// + /// No-op implementation since it is caching. + /// + public void Close() + { + // It's a cached MessageConsumer... + } + + + + + /// + /// Description that shows this is a cached MessageConsumer + /// + /// Description that shows this is a cached MessageConsumer + public override string ToString() + { + return "Cached EMS MessageConsumer: " + this.target; + } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/CachedMessageProducer.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/CachedMessageProducer.cs new file mode 100644 index 00000000..704572e4 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/CachedMessageProducer.cs @@ -0,0 +1,237 @@ +#region License + +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using Spring.Messaging.Ems.Common; +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Connections +{ + /// + /// MessageProducer decorator that adapts calls to a shared MessageProducer + /// instance underneath, managing QoS settings locally within the decorator. + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public class CachedMessageProducer : IMessageProducer + { + private readonly IMessageProducer target; + + private object originalDisableMessageID; + + private object originalDisableMessageTimestamp; + + private MessageDeliveryMode deliveryMode; + + private int priority; + + private long timeToLive; + + /// + /// Initializes a new instance of the class. + /// + /// The target. + public CachedMessageProducer(IMessageProducer target) + { + this.target = target; + deliveryMode = target.MsgDeliveryMode; + priority = target.Priority; + timeToLive = target.TimeToLive; + } + + + /// + /// Gets or sets a value indicating whether disable setting of the message ID property. + /// + /// true if disable message ID setting; otherwise, false. + public bool DisableMessageID + { + get + { + return target.DisableMessageID; + } + set + { + if (originalDisableMessageID == null) + { + originalDisableMessageID = target.DisableMessageID; + } + target.DisableMessageID = value; + } + } + + /// + /// Gets or sets a value indicating whether disable setting the message timestamp property. + /// + /// + /// true if disable message timestamp; otherwise, false. + /// + public bool DisableMessageTimestamp + { + get + { + return target.DisableMessageTimestamp; + } + set + { + if (originalDisableMessageTimestamp == null) + { + originalDisableMessageTimestamp = target.DisableMessageTimestamp; + } + target.DisableMessageTimestamp = value; + } + } + + /// + /// Gets or sets the producer's default delivery mode. + /// + /// The message delivery mode for this message producer + public int DeliveryMode + { + get { return (int)this.deliveryMode; } + set { this.deliveryMode = (MessageDeliveryMode) value; } + } + + /// + /// Gets or sets the MSG delivery mode. + /// + /// The MSG delivery mode. + public MessageDeliveryMode MsgDeliveryMode + { + get { return this.deliveryMode; } + set { this.deliveryMode = value; } + } + + /// + /// Gets or sets the priority of messages sent with this producer. + /// + /// The priority. + public int Priority + { + get { return priority; } + set { priority = value;} + } + + /// + /// Gets or sets the the default length of time in milliseconds from its dispatch time + /// that a produced message should be retained by the message system. + /// + /// Time to live is set to zero by default. + /// The message time to live in milliseconds; zero is unlimited + public long TimeToLive + { + get { return timeToLive; } + set { timeToLive = value; } + } + + public Destination Destination + { + get { return target.Destination; } + } + + /// + /// Gets the target MessageProducer, the producer we are 'wrapping' + /// + /// The target MessageProducer. + public IMessageProducer Target + { + get { return target; } + } + + /// + /// Sends the specified message. + /// + /// The message. + public void Send(Message message) + { + target.Send(message, this.deliveryMode, this.priority, this.timeToLive); + } + + /// + /// Sends the specified message. + /// + /// The message. + /// The delivery mode. + /// The priority. + /// The time to live. + public void Send(Message message, int deliveryMode, int priority, long timeToLive) + { + target.Send(message, deliveryMode, priority, timeToLive); + } + + /// + /// Sends a message to the specified destination. + /// + /// The destination. + /// The message. + public void Send(Destination destination, Message message) + { + target.Send(destination, message, this.deliveryMode, this.priority, this.timeToLive); + } + + public void Send(Message message, MessageDeliveryMode deliveryMode, int priority, long timeToLive) + { + target.Send(message, deliveryMode, priority, timeToLive); + } + + public void Send(Destination dest, Message message, int deliveryMode, int priority, long timeToLive) + { + target.Send(dest, message, deliveryMode, priority, timeToLive); + } + + public void Send(Destination dest, Message message, MessageDeliveryMode deliveryMode, int priority, long timeToLive) + { + target.Send(dest, message, deliveryMode, priority, timeToLive); + } + + /// + /// Reset properties. + /// + public void Close() + { + // It's a cached MessageProducer... reset properties only. + if (originalDisableMessageID != null) + { + target.DisableMessageID = (bool) originalDisableMessageID; + originalDisableMessageID = null; + } + if (originalDisableMessageTimestamp != null) + { + target.DisableMessageTimestamp = (bool) originalDisableMessageTimestamp; + originalDisableMessageTimestamp = null; + } + } + + + + public MessageProducer NativeMessageProducer + { + get { throw new System.NotImplementedException(); } + } + + /// + /// Returns string indicated this is a wrapped MessageProducer + /// + /// + public override string ToString() + { + return "Cached EMS MessageProducer: " + this.target; + } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/CachedSession.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/CachedSession.cs new file mode 100644 index 00000000..3d4ec8bc --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/CachedSession.cs @@ -0,0 +1,678 @@ +#region License + +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using System; +using System.Collections; +using Spring.Messaging.Ems.Common; +using TIBCO.EMS; +using Common.Logging; +using Spring.Collections; +using Spring.Util; +using Queue=TIBCO.EMS.Queue; + +namespace Spring.Messaging.Ems.Connections +{ + /// + /// Wrapper for Session that caches producers and registers itself as available + /// to the session cache when being closed. Generally used for testing purposes or + /// if need to get at the wrapped Session object via the TargetSession property (for + /// vendor specific methods). + /// + /// Juergen Hoeller + /// Mark Pollack + public class CachedSession : IDecoratorSession + { + #region Logging Definition + + private static readonly ILog LOG = LogManager.GetLogger(typeof(CachedSession)); + + #endregion + + private ISession target; + private LinkedList sessionList; + private int sessionCacheSize; + private IDictionary cachedProducers = new Hashtable(); + private IDictionary cachedConsumers = new Hashtable(); + private IMessageProducer cachedUnspecifiedDestinationMessageProducer; + private bool shouldCacheProducers; + private bool shouldCacheConsumers; + private bool transactionOpen = false; + private CachingConnectionFactory ccf; + + /// + /// Initializes a new instance of the class. + /// + /// The target session. + /// The session list. + /// The CachingConnectionFactory. + public CachedSession(ISession targetSession, LinkedList sessionList, CachingConnectionFactory ccf) + { + target = targetSession; + this.sessionList = sessionList; + this.sessionCacheSize = ccf.SessionCacheSize; + shouldCacheProducers = ccf.CacheProducers; + shouldCacheConsumers = ccf.CacheConsumers; + this.ccf = ccf; + } + + + /// + /// Gets the target, for testing purposes. + /// + /// The target. + public ISession TargetSession + { + get { return target; } + } + + /// + /// Creates the producer, potentially returning a cached instance. + /// + /// The destination. + /// A message producer. + public IMessageProducer CreateProducer(Destination destination) + { + if (shouldCacheProducers) + { + IMessageProducer producer = (IMessageProducer)cachedProducers[destination]; + if (producer != null) + { + #region Logging + + if (LOG.IsDebugEnabled) + { + LOG.Debug("Found cached MessageProducer for destination [" + destination + "]"); + } + + #endregion + } + else + { + producer = target.CreateProducer(destination); + #region Logging + + if (LOG.IsDebugEnabled) + { + LOG.Debug("Creating cached MessageProducer for destination [" + destination + "]"); + } + + #endregion + cachedProducers.Add(destination, producer); + + } + this.transactionOpen = true; + return new CachedMessageProducer(producer); + } + else + { + return target.CreateProducer(destination); + } + } + + + /// + /// If have not yet reached session cache size, cache the session, otherwise + /// dispose of all cached message producers and close the session. + /// + public void Close() + { + if (ccf.IsActive) + { + //don't pass the call to the underlying target. + lock (sessionList) + { + if (sessionList.Count < sessionCacheSize) + { + LogicalClose(); + // Remain open in the session list. + return; + } + } + } + // If we get here, we're supposed to shut down. + PhysicalClose(); + } + + private void LogicalClose() + { + // Preserve rollback-on-close semantics. + if (this.transactionOpen && this.target.Transacted) + { + this.transactionOpen = false; + this.target.Rollback(); + } + + // Physically close durable subscribers at time of Session close call. + IList ToRemove = new ArrayList(); + foreach (DictionaryEntry dictionaryEntry in cachedConsumers) + { + ConsumerCacheKey key = (ConsumerCacheKey) dictionaryEntry.Key; + if (key.Subscription != null) + { + ((IMessageConsumer) dictionaryEntry.Value).Close(); + ToRemove.Add(key); + } + } + foreach (ConsumerCacheKey key in ToRemove) + { + cachedConsumers.Remove(key); + } + + // Allow for multiple close calls... + if (!sessionList.Contains(this)) + { + #region Logging + + if (LOG.IsDebugEnabled) + { + LOG.Debug("Returning cached Session: " + target); + } + + #endregion + + sessionList.Add(this); //add to end of linked list. + } + } + + private void PhysicalClose() + { + if (LOG.IsDebugEnabled) + { + LOG.Debug("Closing cached Session: " + this.target); + } + // Explicitly close all MessageProducers and MessageConsumers that + // this Session happens to cache... + try + { + foreach (DictionaryEntry entry in cachedProducers) + { + ((IMessageProducer)entry.Value).Close(); + } + foreach (DictionaryEntry entry in cachedConsumers) + { + ((IMessageConsumer)entry.Value).Close(); + } + } + finally + { + // Now actually close the Session. + target.Close(); + } + } + + /// + /// Creates the consumer, potentially returning a cached instance. + /// + /// The destination. + /// A message consumer + public IMessageConsumer CreateConsumer(Destination destination) + { + return CreateConsumer(destination, null, false, null); + } + + + /// + /// Creates the consumer, potentially returning a cached instance. + /// + /// The destination. + /// The selector. + /// A message consumer + public IMessageConsumer CreateConsumer(Destination destination, string selector) + { + return CreateConsumer(destination, selector, false, null); + } + + /// + /// Creates the consumer, potentially returning a cached instance. + /// + /// The destination. + /// The selector. + /// if set to true [no local]. + /// A message consumer. + public IMessageConsumer CreateConsumer(Destination destination, string selector, bool noLocal) + { + return CreateConsumer(destination, selector, noLocal, null); + } + + + /// + /// Creates the durable consumer, potentially returning a cached instance. + /// + /// The destination. + /// The name of the durable subscription. + /// The selector. + /// if set to true [no local]. + /// A message consumer + public ITopicSubscriber CreateDurableSubscriber(Topic destination, string subscription, string selector, bool noLocal) + { + this.transactionOpen = true; + if (shouldCacheConsumers) + { + return (ITopicSubscriber)GetCachedConsumer(destination, selector, noLocal, subscription); + } + else + { + return target.CreateDurableSubscriber(destination, subscription, selector, noLocal); + } + } + + /// + /// Creates the durable consumer, potentially returning a cached instance. + /// + /// The destination. + /// The name of the durable subscription. + /// A message consumer + public ITopicSubscriber CreateDurableSubscriber(Topic destination, string subscription) + { + return CreateDurableSubscriber(destination, subscription, null, false); + } + + + + /// + /// Deletes the durable consumer. + /// + /// The name. + public void DeleteDurableConsumer(string name) + { + throw new NotImplementedException(); + } + + /// + /// Deletes the durable consumer. + /// + /// The name. + /// The request timeout. + public void DeleteDurableConsumer(string name, TimeSpan requestTimeout) + { + throw new NotImplementedException(); + } + + /// + /// Creates the consumer. + /// + /// The destination. + /// The selector. + /// if set to true [no local]. + /// The subscription. + /// + protected IMessageConsumer CreateConsumer(Destination destination, string selector, bool noLocal, string subscription) + { + this.transactionOpen = true; + if (shouldCacheConsumers) + { + return GetCachedConsumer(destination, selector, noLocal, subscription); + } + else + { + return target.CreateConsumer(destination, selector, noLocal); + } + } + + private IMessageConsumer GetCachedConsumer(Destination destination, string selector, bool noLocal, string subscription) + { + object cacheKey = new ConsumerCacheKey(destination, selector, noLocal, null); + IMessageConsumer consumer = (IMessageConsumer)cachedConsumers[cacheKey]; + if (consumer != null) + { + if (LOG.IsDebugEnabled) + { + LOG.Debug("Found cached EMS MessageConsumer for destination [" + destination + "]: " + consumer); + } + } + else + { + if (destination is Topic) + { + consumer = (subscription != null + ? target.CreateDurableSubscriber((Topic)destination, subscription, selector, noLocal) + : target.CreateConsumer(destination, selector, noLocal)); + } + else + { + consumer = target.CreateConsumer(destination, selector); + } + if (LOG.IsDebugEnabled) + { + LOG.Debug("Creating cached NMS MessageConsumer for destination [" + destination + "]: " + consumer); + } + cachedConsumers[cacheKey] = consumer; + } + return new CachedMessageConsumer(consumer); + } + + #region Pass through implementations + + /// + /// Gets the queue. + /// + /// The name. + /// + public TIBCO.EMS.Queue CreateQueue(string name) + { + this.transactionOpen = true; + return target.CreateQueue(name); + } + + /// + /// Gets the topic. + /// + /// The name. + /// + public Topic CreateTopic(string name) + { + this.transactionOpen = true; + return target.CreateTopic(name); + } + + /// + /// Creates the temporary queue. + /// + /// + public TemporaryQueue CreateTemporaryQueue() + { + this.transactionOpen = true; + return target.CreateTemporaryQueue(); + } + + /// + /// Creates the temporary topic. + /// + /// + public TemporaryTopic CreateTemporaryTopic() + { + this.transactionOpen = true; + return target.CreateTemporaryTopic(); + } + + /// + /// Creates the message. + /// + /// + public Message CreateMessage() + { + this.transactionOpen = true; + return target.CreateMessage(); + } + + /// + /// Creates the text message. + /// + /// + public TextMessage CreateTextMessage() + { + this.transactionOpen = true; + return target.CreateTextMessage(); + } + + /// + /// Creates the text message. + /// + /// The text. + /// + public TextMessage CreateTextMessage(string text) + { + this.transactionOpen = true; + return target.CreateTextMessage(text); + } + + /// + /// Creates the map message. + /// + /// + public MapMessage CreateMapMessage() + { + this.transactionOpen = true; + return target.CreateMapMessage(); + } + + /// + /// Creates the bytes message. + /// + /// + public BytesMessage CreateBytesMessage() + { + this.transactionOpen = true; + return target.CreateBytesMessage(); + } + + /// + /// Creates the object message. + /// + /// + public ObjectMessage CreateObjectMessage() + { + this.transactionOpen = true; + return target.CreateObjectMessage(); + } + + /// + /// Creates the object message. + /// + /// The body. + /// + public ObjectMessage CreateObjectMessage(object body) + { + this.transactionOpen = true; + return target.CreateObjectMessage(body); + } + + /// + /// Creates the stream message. + /// + /// + public StreamMessage CreateStreamMessage() + { + this.transactionOpen = true; + return target.CreateStreamMessage(); + } + + + /// + /// Commits this instance. + /// + public void Commit() + { + this.transactionOpen = false; + target.Commit(); + } + + /// + /// Rollbacks this instance. + /// + public void Rollback() + { + this.transactionOpen = false; + target.Rollback(); + } + + + public QueueBrowser CreateBrowser(Queue queue) + { + this.transactionOpen = true; + return target.CreateBrowser(queue); + } + + public QueueBrowser CreateBrowser(Queue queue, string messageSelector) + { + this.transactionOpen = true; + return target.CreateBrowser(queue, messageSelector); + } + + + public void Recover() + { + this.transactionOpen = true; + target.Recover(); + } + + public void Run() + { + this.transactionOpen = true; + target.Run(); + } + + public void Unsubscribe(string name) + { + this.transactionOpen = true; + target.Unsubscribe(name); + } + + /// + /// Gets a value indicating whether this is transacted. + /// + /// true if transacted; otherwise, false. + public bool Transacted + { + get + { + this.transactionOpen = true; + return target.Transacted; + } + } + + /// + /// Gets the acknowledgement mode. + /// + /// The acknowledgement mode. + public SessionMode SessionAcknowledgeMode + { + get + { + this.transactionOpen = true; + return target.SessionAcknowledgeMode; + } + } + + public long SessID + { + get + { + this.transactionOpen = true; + return target.SessID; + } + } + + public Session NativeSession + { + get + { + this.transactionOpen = true; + return target.NativeSession; + } + } + + public int AcknowledgeMode + { + get + { + this.transactionOpen = true; + return target.AcknowledgeMode; + } + } + + public Connection Connection + { + get + { + this.transactionOpen = true; + return target.Connection; + } + } + + public bool IsClosed + { + get { + this.transactionOpen = true; + return target.IsClosed; + } + } + + public bool IsTransacted + { + get + { + this.transactionOpen = true; + return target.IsTransacted; + } + } + + public IMessageListener MessageListener + { + get { throw new System.NotImplementedException(); } + set { throw new System.NotImplementedException(); } + } + + #endregion + + /// + /// Returns a that represents the current . + /// + /// + /// A that represents the current . + /// + public override string ToString() + { + return "Cached EMS Session: " + this.target; + } + } + + internal class ConsumerCacheKey + { + private Destination destination; + private string selector; + private bool noLocal; + private string subscription; + + public ConsumerCacheKey(Destination destination, string selector, bool noLocal, string subscription) + { + this.destination = destination; + this.selector = selector; + this.noLocal = noLocal; + this.subscription = subscription; + } + + public string Subscription + { + get { return subscription; } + } + + protected bool Equals(ConsumerCacheKey consumerCacheKey) + { + if (consumerCacheKey == null) return false; + if (!Equals(destination, consumerCacheKey.destination)) return false; + if (!ObjectUtils.NullSafeEquals(selector, consumerCacheKey.selector)) return false; + if (!Equals(noLocal, consumerCacheKey.noLocal)) return false; + if (!ObjectUtils.NullSafeEquals(subscription, consumerCacheKey.subscription)) return false; + return true; + } + + public override bool Equals(object obj) + { + if (ReferenceEquals(this, obj)) return true; + return Equals(obj as ConsumerCacheKey); + } + + public override int GetHashCode() + { + return destination.GetHashCode(); + } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/CachingConnectionFactory.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/CachingConnectionFactory.cs new file mode 100644 index 00000000..9ba643df --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/CachingConnectionFactory.cs @@ -0,0 +1,278 @@ +#region License + +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using System; +using System.Collections; +using Spring.Messaging.Ems.Common; +using TIBCO.EMS; +using Common.Logging; +using Spring.Collections; +using Spring.Util; + +namespace Spring.Messaging.Ems.Connections +{ + /// + /// subclass that adds + /// Session, MessageProducer, and MessageConsumer caching. This ConnectionFactory + /// also switches the ReconnectOnException property to true + /// by default, allowing for automatic recovery of the underlying + /// Connection. + /// + /// + /// By default, only one single Session will be cached, with further requested + /// Sessions being created and disposed on demand. Consider raising the + /// SessionCacheSize property in case of a high-concurrency environment. + /// + /// NOTE: This ConnectionFactory requires explicit closing of all Sessions + /// obtained from its shared Connection. This is the usual recommendation for + /// native NMS access code anyway. However, with this ConnectionFactory, its use + /// is mandatory in order to actually allow for Session reuse. + /// + /// + /// Note also that MessageConsumers obtained from a cached Session won't get + /// closed until the Session will eventually be removed from the pool. This may + /// lead to semantic side effects in some cases. For a durable subscriber, the + /// logical Session.Close() call will also close the subscription. + /// Re-registering a durable consumer for the same subscription on the same + /// Session handle is not supported; close and reobtain a cached Session first. + /// + /// + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public class CachingConnectionFactory : SingleConnectionFactory + { + #region Logging Definition + + private static readonly ILog LOG = LogManager.GetLogger(typeof(CachingConnectionFactory)); + + #endregion + + private int sessionCacheSize = 1; + + private bool cacheProducers = true; + + private bool cacheConsumers = true; + + private volatile bool active = true; + + private IDictionary cachedSessions = new Hashtable(); + + + /// + /// Initializes a new instance of the class. + /// and sets the ReconnectOnException to true + /// + public CachingConnectionFactory() + { + ReconnectOnException = true; + } + + /// + /// Initializes a new instance of the class for the given + /// IConnectionFactory + /// + /// The target connection factory. + public CachingConnectionFactory(IConnectionFactory targetConnectionFactory) : base(targetConnectionFactory) + { + ReconnectOnException = true; + } + + + /// + /// Gets or sets the size of the session cache. + /// + /// + /// This cache size is the maximum limit for the number of cached Sessions + /// per session acknowledgement type (auto, client, dups_ok, transacted). + /// As a consequence, the actual number of cached Sessions may be up to + /// four times as high as the specified value - in the unlikely case + /// of mixing and matching different acknowledgement types. + /// + /// Default is 1: caching a single Session, (re-)creating further ones on + /// demand. Specify a number like 10 if you'd like to raise the number of cached + /// Sessions; that said, 1 may be sufficient for low-concurrency scenarios. + /// + /// + /// The size of the session cache. + public int SessionCacheSize + { + get { return sessionCacheSize; } + set + { + AssertUtils.IsTrue(value >= 1, "Session cache size must be 1 or higher"); + sessionCacheSize = value; + } + } + + + /// + /// Gets or sets a value indicating whether to cache MessageProducers per + /// Session instance. (more specifically: one MessageProducer per Destination + /// and Session). + /// + /// + /// Default is "true". Switch this to "false" in order to always, + /// recreate MessageProducers on demand. + /// + /// + /// true if should cache message producers; otherwise, false. + public bool CacheProducers + { + get { return cacheProducers; } + set { cacheProducers = value; } + } + + + /// + /// Gets or sets a value indicating whether o cache JMS MessageConsumers per + /// NMS Session instance. + /// + /// + /// Mmore specifically: one MessageConsumer per Destination, selector String + /// and Session. Note that durable subscribers will only be cached until + /// logical closing of the Session handle. + /// + /// Default is "true". Switch this to "false" in order to always + /// recreate MessageConsumers on demand. + /// + /// + /// true to cache consumers per session instance; otherwise, false. + public bool CacheConsumers + { + get { return cacheConsumers; } + set { cacheConsumers = value; } + } + + /// + /// Gets or sets a value indicating whether this instance is active. + /// + /// true if this instance is active; otherwise, false. + public bool IsActive + { + get { return active; } + set { active = value; } + } + + /// + /// Resets the Session cache as well as resetting the connection. + /// + public override void ResetConnection() + { + this.active = false; + lock (cachedSessions) + { + foreach (DictionaryEntry dictionaryEntry in cachedSessions) + { + LinkedList sessionList = (LinkedList) dictionaryEntry.Value; + lock (sessionList) + { + foreach (ISession session in sessionList) + { + try + { + session.Close(); + } + catch (Exception ex) + { + LOG.Trace("Could not close cached NMS Session", ex); + } + } + } + } + cachedSessions.Clear(); + } + this.active = true; + // Now proceed with actual closing of the shared Connection... + base.ResetConnection(); + } + + /// + /// Obtaining a cached Session. + /// + /// The connection to operate on. + /// The session ack mode. + /// The Session to use + /// + public override ISession GetSession(IConnection con, SessionMode mode) + { + LinkedList sessionList; + lock (cachedSessions) + { + sessionList = (LinkedList) cachedSessions[mode]; + if (sessionList == null) + { + sessionList = new LinkedList(); + cachedSessions.Add(mode, sessionList); + } + } + + ISession session = null; + lock (sessionList) + { + if (sessionList.Count > 0) + { + session = (ISession) sessionList[0]; + sessionList.RemoveAt(0); + } + } + if (session != null) + { + if (LOG.IsDebugEnabled) + { + LOG.Debug("Found cached Session for mode " + mode + ": " + + (session is IDecoratorSession ? ((IDecoratorSession) session).TargetSession : session)); + } + } else + { + ISession targetSession = CreateSession(con, mode); + if (LOG.IsDebugEnabled) + { + LOG.Debug("Creating cached Session for mode " + mode + ": " + targetSession); + } + session = GetCachedSessionWrapper(targetSession, sessionList); + } + return session; + } + + protected virtual ISession CreateSession(IConnection con, SessionMode mode) + { + bool transacted = (mode == SessionMode.SessionTransacted); + SessionMode ackMode = (transacted ? SessionMode.SessionTransacted : mode); + return con.CreateSession(transacted, ackMode); + } + + + /// + /// Wraps the given Session so that it delegates every method call to the target session but + /// adapts close calls. This is useful for allowing application code to + /// handle a special framework Session just like an ordinary Session. + /// + /// The original Session to wrap. + /// The List of cached Sessions that the given Session belongs to. + /// The wrapped Session + protected virtual ISession GetCachedSessionWrapper(ISession targetSession, LinkedList sessionList) + { + return new CachedSession(targetSession, sessionList, this); + } + } + + +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ConnectionFactoryUtils.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ConnectionFactoryUtils.cs index 6dc6aa44..29adc486 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ConnectionFactoryUtils.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ConnectionFactoryUtils.cs @@ -20,6 +20,7 @@ using System; using Common.Logging; +using Spring.Messaging.Ems.Common; using Spring.Messaging.Ems.Support; using Spring.Transaction.Support; using Spring.Util; @@ -41,25 +42,59 @@ namespace Spring.Messaging.Ems.Connections #endregion /// - /// Releases the given connection. + /// Releases the given connection, stopping it (if necessary) and eventually closing it. /// + /// Checks , if available. + /// This is essentially a more sophisticated version of + /// + /// /// The connection to release. (if this is null, the call will be ignored) /// The ConnectionFactory that the Connection was obtained from. (may be null) /// whether the Connection might have been started by the application. - public static void ReleaseConnection(Connection connection, ConnectionFactory cf, bool started) + public static void ReleaseConnection(IConnection connection, IConnectionFactory cf, bool started) { if (connection == null) { return; } + if (started && cf is ISmartConnectionFactory && ((ISmartConnectionFactory)cf).ShouldStop(connection)) + { + try + { + connection.Stop(); + } + catch (Exception ex) + { + LOG.Debug("Could not stop EMS Connection before closing it", ex); + + } + } try { connection.Close(); - } catch (Exception ex) + } + catch (Exception ex) { LOG.Debug("Could not close EMS Connection", ex); - } + } + } + + /// + /// Return the innermost target Session of the given Session. If the given + /// Session is a decorated session, it will be unwrapped until a non-decorated + /// Session is found. Otherwise, the passed-in Session will be returned as-is. + /// + /// The session to unwrap + /// The innermost target Session, or the passed-in one if no decorator + public static ISession GetTargetSession(ISession session) + { + ISession sessionToUse = session; + while (sessionToUse is IDecoratorSession) + { + sessionToUse = ((IDecoratorSession)sessionToUse).TargetSession; + } + return sessionToUse; } /// @@ -71,7 +106,7 @@ namespace Spring.Messaging.Ems.Connections /// /// true if is session transactional, bound to current thread; otherwise, false. /// - public static bool IsSessionTransactional(Session session, ConnectionFactory cf) + public static bool IsSessionTransactional(ISession session, IConnectionFactory cf) { if (session == null || cf == null) { @@ -96,7 +131,7 @@ namespace Spring.Messaging.Ems.Connections /// the transactional Session, or null if none found /// /// EMSException in case of EMS failure - public static Session GetTransactionalSession(ConnectionFactory cf, Connection existingCon, + public static ISession GetTransactionalSession(IConnectionFactory cf, IConnection existingCon, bool synchedLocalTransactionAllowed) { return @@ -119,7 +154,7 @@ namespace Spring.Messaging.Ems.Connections /// the transactional Session, or null if none found /// /// EMSException in case of EMS failure - public static Session DoGetTransactionalSession(Object resourceKey, ResourceFactory resourceFactory, bool startConnection) + public static ISession DoGetTransactionalSession(Object resourceKey, ResourceFactory resourceFactory, bool startConnection) { AssertUtils.ArgumentNotNull(resourceKey, "Resource key must not be null"); AssertUtils.ArgumentNotNull(resourceKey, "ResourceFactory must not be null"); @@ -128,12 +163,12 @@ namespace Spring.Messaging.Ems.Connections (EmsResourceHolder)TransactionSynchronizationManager.GetResource(resourceKey); if (resourceHolder != null) { - Session rhSession = resourceFactory.GetSession(resourceHolder); + ISession rhSession = resourceFactory.GetSession(resourceHolder); if (rhSession != null) { if (startConnection) { - Connection conn = resourceFactory.GetConnection(resourceHolder); + IConnection conn = resourceFactory.GetConnection(resourceHolder); if (conn != null) { conn.Start(); @@ -152,8 +187,8 @@ namespace Spring.Messaging.Ems.Connections resourceHolderToUse = new EmsResourceHolder(); } - Connection con = resourceFactory.GetConnection(resourceHolderToUse); - Session session = null; + IConnection con = resourceFactory.GetConnection(resourceHolderToUse); + ISession session = null; try { bool isExistingCon = (con != null); @@ -210,39 +245,39 @@ namespace Spring.Messaging.Ems.Connections private class AnonymousClassResourceFactory : ResourceFactory { - private Connection existingCon; - private ConnectionFactory cf; + private IConnection existingCon; + private IConnectionFactory cf; private bool synchedLocalTransactionAllowed; - public AnonymousClassResourceFactory(Connection existingCon, ConnectionFactory cf, + public AnonymousClassResourceFactory(IConnection existingCon, IConnectionFactory cf, bool synchedLocalTransactionAllowed) { InitBlock(existingCon, cf, synchedLocalTransactionAllowed); } - private void InitBlock(Connection existingCon, ConnectionFactory cf, bool synchedLocalTransactionAllowed) + private void InitBlock(IConnection existingCon, IConnectionFactory cf, bool synchedLocalTransactionAllowed) { this.existingCon = existingCon; this.cf = cf; this.synchedLocalTransactionAllowed = synchedLocalTransactionAllowed; } - public virtual Session GetSession(EmsResourceHolder holder) + public virtual ISession GetSession(EmsResourceHolder holder) { - return holder.GetSession(typeof(Session), existingCon); + return holder.GetSession(typeof(ISession), existingCon); } - public virtual Connection GetConnection(EmsResourceHolder holder) + public virtual IConnection GetConnection(EmsResourceHolder holder) { return (existingCon != null ? existingCon : holder.GetConnection()); } - public virtual Connection CreateConnection() + public virtual IConnection CreateConnection() { return cf.CreateConnection(); } - public virtual Session CreateSession(Connection con) + public virtual ISession CreateSession(IConnection con) { return con.CreateSession(synchedLocalTransactionAllowed, Session.SESSION_TRANSACTED); } @@ -268,7 +303,7 @@ namespace Spring.Messaging.Ems.Connections /// an appropriate Session fetched from the holder, /// or null if none found /// - Session GetSession(EmsResourceHolder holder); + ISession GetSession(EmsResourceHolder holder); /// Fetch an appropriate Connection from the given EmsResourceHolder. /// the EmsResourceHolder @@ -276,13 +311,13 @@ namespace Spring.Messaging.Ems.Connections /// an appropriate Connection fetched from the holder, /// or null if none found /// - Connection GetConnection(EmsResourceHolder holder); + IConnection GetConnection(EmsResourceHolder holder); /// Create a new EMS Connection for registration with a EmsResourceHolder. /// the new EMS Connection /// /// EMSException if thrown by EMS API methods - Connection CreateConnection(); + IConnection CreateConnection(); /// Create a new EMS Session for registration with a EmsResourceHolder. /// the EMS Connection to create a Session for @@ -290,7 +325,7 @@ namespace Spring.Messaging.Ems.Connections /// the new EMS Session /// /// EMSException if thrown by EMS API methods - Session CreateSession(Connection con); + ISession CreateSession(IConnection con); /// diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsResourceHolder.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsResourceHolder.cs index 2335c340..0a7cfdb2 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsResourceHolder.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsResourceHolder.cs @@ -22,6 +22,7 @@ using System; using System.Collections; using Common.Logging; using Spring.Collections; +using Spring.Messaging.Ems.Common; using Spring.Transaction.Support; using Spring.Util; using TIBCO.EMS; @@ -47,7 +48,7 @@ namespace Spring.Messaging.Ems.Connections #region Fields - private ConnectionFactory connectionFactory; + private IConnectionFactory connectionFactory; private bool frozen = false; @@ -75,7 +76,7 @@ namespace Spring.Messaging.Ems.Connections /// The connection factory that this /// resource holder is associated with (may be null) /// - public EmsResourceHolder(ConnectionFactory connectionFactory) + public EmsResourceHolder(IConnectionFactory connectionFactory) { this.connectionFactory = connectionFactory; } @@ -85,7 +86,7 @@ namespace Spring.Messaging.Ems.Connections /// given Session. /// /// The session. - public EmsResourceHolder(Session session) + public EmsResourceHolder(ISession session) { AddSession(session); frozen = true; @@ -96,7 +97,7 @@ namespace Spring.Messaging.Ems.Connections /// /// the EMS Session /// - public EmsResourceHolder(Connection connection, Session session) + public EmsResourceHolder(IConnection connection, ISession session) { AddConnection(connection); AddSession(session, connection); @@ -109,7 +110,7 @@ namespace Spring.Messaging.Ems.Connections /// The connection factory. /// The connection. /// The session. - public EmsResourceHolder(ConnectionFactory connectionFactory, Connection connection, Session session) + public EmsResourceHolder(IConnectionFactory connectionFactory, IConnection connection, ISession session) { this.connectionFactory = connectionFactory; AddConnection(connection); @@ -142,7 +143,7 @@ namespace Spring.Messaging.Ems.Connections /// Adds the connection to the list of resources managed by this holder. /// /// The connection. - public void AddConnection(Connection connection) + public void AddConnection(IConnection connection) { AssertUtils.IsTrue(!frozen, "Cannot add Connection because EmsResourceHolder is frozen"); AssertUtils.ArgumentNotNull(connection, "Connection must not be null"); @@ -156,7 +157,7 @@ namespace Spring.Messaging.Ems.Connections /// Adds the session to the list of resources managed by this holder. /// /// The session. - public void AddSession(Session session) + public void AddSession(ISession session) { AddSession(session, null); } @@ -166,7 +167,7 @@ namespace Spring.Messaging.Ems.Connections /// /// The session. /// The connection. - public void AddSession(Session session, Connection connection) + public void AddSession(ISession session, IConnection connection) { AssertUtils.IsTrue(!frozen, "Cannot add Session because EmsResourceHolder is frozen"); AssertUtils.ArgumentNotNull(session, "Session must not be null"); @@ -190,9 +191,9 @@ namespace Spring.Messaging.Ems.Connections /// Gets the connection managed by this resource holder /// /// A Connection, or null if no managed connection. - public virtual Connection GetConnection() + public virtual IConnection GetConnection() { - return (!(this.connections.Count == 0) ? (Connection)this.connections[0] : null); + return (!(this.connections.Count == 0) ? (IConnection)this.connections[0] : null); } /// @@ -202,18 +203,18 @@ namespace Spring.Messaging.Ems.Connections /// /// Type of the connection. /// The connection, or null if not found. - public virtual Connection GetConnection(Type connectionType) + public virtual IConnection GetConnection(Type connectionType) { - return (Connection)CollectionUtils.FindValueOfType(this.connections, connectionType); + return (IConnection)CollectionUtils.FindValueOfType(this.connections, connectionType); } /// /// Gets the first session manged by this holder or null if not available. /// /// The session or null if not available. - public virtual Session GetSession() + public virtual ISession GetSession() { - return (!(this.sessions.Count == 0) ? (Session)this.sessions[0] : null); + return (!(this.sessions.Count == 0) ? (ISession)this.sessions[0] : null); } /// @@ -221,7 +222,7 @@ namespace Spring.Messaging.Ems.Connections /// /// Type of the session. /// The session or null if not available. - public virtual Session GetSession(Type sessionType) + public virtual ISession GetSession(Type sessionType) { return GetSession(sessionType, null); } @@ -232,14 +233,14 @@ namespace Spring.Messaging.Ems.Connections /// Type of the session. /// The connection. /// The sessin or null if not available. - public virtual Session GetSession(Type sessionType, Connection connection) + public virtual ISession GetSession(Type sessionType, IConnection connection) { IList sessions = this.sessions; if (connection != null) { sessions = (IList)sessionsPerConnection[connection]; } - return (Session)CollectionUtils.FindValueOfType(sessions, sessionType); + return (ISession)CollectionUtils.FindValueOfType(sessions, sessionType); } /// @@ -247,7 +248,7 @@ namespace Spring.Messaging.Ems.Connections /// public virtual void CommitAll() { - foreach (Session session in sessions) + foreach (ISession session in sessions) { session.Commit(); } @@ -258,7 +259,7 @@ namespace Spring.Messaging.Ems.Connections /// public virtual void CloseAll() { - foreach (Session session in sessions) + foreach (ISession session in sessions) { try { @@ -269,7 +270,7 @@ namespace Spring.Messaging.Ems.Connections logger.Debug("Could not close EMS Session after transaction", ex); } } - foreach (Connection connection in connections) + foreach (IConnection connection in connections) { ConnectionFactoryUtils.ReleaseConnection(connection, connectionFactory, true); } @@ -285,7 +286,7 @@ namespace Spring.Messaging.Ems.Connections /// /// true if the holder contains the specified session; otherwise, false. /// - public bool ContainsSession(Session session) + public bool ContainsSession(ISession session) { return this.sessions.Contains(session); } diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsTransactionManager.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsTransactionManager.cs index 2b1b6f3f..15b8238c 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsTransactionManager.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/EmsTransactionManager.cs @@ -21,6 +21,7 @@ using System; using System.Data; using Common.Logging; +using Spring.Messaging.Ems.Common; using Spring.Messaging.Ems.Core; using Spring.Objects.Factory; using Spring.Transaction; @@ -60,7 +61,7 @@ namespace Spring.Messaging.Ems.Connections #endregion - private ConnectionFactory connectionFactory; + private IConnectionFactory connectionFactory; /// /// Initializes a new instance of the class. @@ -86,7 +87,7 @@ namespace Spring.Messaging.Ems.Connections /// given a ConnectionFactory. /// /// The connection factory to obtain connections from. - public EmsTransactionManager(ConnectionFactory connectionFactory) : this() + public EmsTransactionManager(IConnectionFactory connectionFactory) : this() { ConnectionFactory = connectionFactory; AfterPropertiesSet(); @@ -98,7 +99,7 @@ namespace Spring.Messaging.Ems.Connections /// for. /// /// The connection factory. - public ConnectionFactory ConnectionFactory + public IConnectionFactory ConnectionFactory { get { return connectionFactory; } set @@ -173,8 +174,8 @@ namespace Spring.Messaging.Ems.Connections throw new InvalidIsolationLevelException("EMS does not support an isoliation level concept"); } EmsTransactionObject txObject = (EmsTransactionObject) transaction; - Connection con = null; - Session session = null; + IConnection con = null; + ISession session = null; try { con = CreateConnection(); @@ -266,7 +267,7 @@ namespace Spring.Messaging.Ems.Connections protected override void DoCommit(DefaultTransactionStatus status) { EmsTransactionObject txObject = (EmsTransactionObject)status.Transaction; - Session session = txObject.ResourceHolder.GetSession(); + ISession session = txObject.ResourceHolder.GetSession(); try { if (status.Debug) @@ -295,7 +296,7 @@ namespace Spring.Messaging.Ems.Connections protected override void DoRollback(DefaultTransactionStatus status) { EmsTransactionObject txObject = (EmsTransactionObject)status.Transaction; - Session session = txObject.ResourceHolder.GetSession(); + ISession session = txObject.ResourceHolder.GetSession(); try { if (status.Debug) @@ -373,7 +374,7 @@ namespace Spring.Messaging.Ems.Connections /// /// The new Connection /// If thrown by underlying messaging APIs - protected virtual Connection CreateConnection() + protected virtual IConnection CreateConnection() { return ConnectionFactory.CreateConnection(); } @@ -384,7 +385,7 @@ namespace Spring.Messaging.Ems.Connections /// The connection to create a Session for. /// the new Session /// If thrown by underlying messaging APIs - protected virtual Session CreateSession(Connection connection) + protected virtual ISession CreateSession(IConnection connection) { return connection.CreateSession(true, Session.SESSION_TRANSACTED); } diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/IDecoratorSession.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/IDecoratorSession.cs new file mode 100644 index 00000000..1f3b3f1b --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/IDecoratorSession.cs @@ -0,0 +1,42 @@ +#region License + +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using Spring.Messaging.Ems.Common; + +namespace Spring.Messaging.Ems.Connections +{ + /// + /// Subinterface of Session to be implemented by + /// implementations that wrap an Session to provide added + /// functionality. Allows access to the the underlying target Session. + /// + /// Mark Pollack + /// + /// + public interface IDecoratorSession : ISession + { + /// + /// Gets the target session of the decorator. + /// This will typically be the native provider Session or a wrapper from a session pool. + /// + /// The underlying session, never null + ISession TargetSession { get; } + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ISmartConnectionFactory.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ISmartConnectionFactory.cs new file mode 100644 index 00000000..526e189e --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/ISmartConnectionFactory.cs @@ -0,0 +1,43 @@ + + +#region License + +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using Spring.Messaging.Ems.Common; +using TIBCO.EMS; + +namespace Spring.Messaging.Ems.Connections +{ + /// + /// Extension of the IConnectionFactory interface, + /// indicating how to release Connections obtained from it. + /// + /// Juergen Hoeller + /// Mark Pollack (.NET) + public interface ISmartConnectionFactory + { + /// + /// Shoulds we stop the connection, obtained from this ConnectionFactory? + /// + /// The connection to check. + /// wheter a stop call is necessary + bool ShouldStop(IConnection con); + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/SingleConnectionFactory.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/SingleConnectionFactory.cs new file mode 100644 index 00000000..9f2c3556 --- /dev/null +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Connections/SingleConnectionFactory.cs @@ -0,0 +1,777 @@ +#region License + +/* + * Copyright © 2002-2008 the original author or authors. + * + * Licensed 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. + */ + +#endregion + +using System; +using System.IO; +using System.Runtime.Serialization; +using Spring.Messaging.Ems.Support; +using TIBCO.EMS; +using Common.Logging; +using Spring.Messaging.Ems.Core; +using Spring.Messaging.Ems.Common; +using Spring.Objects.Factory; +using Spring.Util; + +namespace Spring.Messaging.Ems.Connections +{ + /// + /// A ConnectionFactory adapter that returns the same Connection + /// from all CreateConnection() calls, and ignores calls to + /// Connection.Close(). According to the JMS Connection + /// model, this is perfectly thread-safe. The + /// shared Connection can be automatically recovered in case of an Exception. + /// + /// + /// + /// You can either pass in a specific Connection directly or let this + /// factory lazily create a Connection via a given target ConnectionFactory. + /// + /// + /// Useful for testing and in applications when you want to keep using the + /// same Connection for multiple + /// calls, without having a pooling ConnectionFactory underneath. This may span + /// any number of transactions, even concurrently executing transactions. + /// + /// + /// Note that Spring's message listener containers support the use of + /// a shared Connection within each listener container instance. Using + /// SingleConnectionFactory with a MessageListenerContainer only really makes sense for + /// sharing a single Connection across multiple listener containers. + /// + /// + /// Juergen Hoeller + /// Mark Pollack + /// Mark Pollack (.NET) + public class SingleConnectionFactory : IConnectionFactory, IExceptionListener, IInitializingObject, IDisposable + { + #region Logging Definition + + private static readonly ILog LOG = LogManager.GetLogger(typeof (SingleConnectionFactory)); + + #endregion + + #region Fields + + private IConnectionFactory targetConnectionFactory; + + private string clientId; + + private IExceptionListener exceptionListener; + + private bool reconnectOnException = false; + + /// + /// Wrapped Connection + /// + private IConnection target; + + /// + /// Proxy Connection + /// + private IConnection connection; + + + /// + /// Whether the shared Connection has been started + /// + private bool started = false; + + /// + /// Synchronization monitor for the shared Connection + /// + private object connectionMonitor = new object(); + + + #endregion + + #region Constructors + + /// + /// Initializes a new instance of the class. + /// + public SingleConnectionFactory() + { + } + + + /// + /// Initializes a new instance of the class + /// that alwasy returns the given Connection. + /// + /// The single Connection. + public SingleConnectionFactory(IConnection target) + { + AssertUtils.ArgumentNotNull(target, "connection", "TargetSession Connection must not be null"); + this.target = target; + connection = GetSharedConnection(target); + } + + + /// + /// Initializes a new instance of the class + /// that alwasy returns a single Connection. + /// + /// The target connection factory. + public SingleConnectionFactory(IConnectionFactory targetConnectionFactory) + { + AssertUtils.ArgumentNotNull(targetConnectionFactory, "targetConnectionFactory", + "TargetSession ConnectionFactory must not be null"); + this.targetConnectionFactory = targetConnectionFactory; + } + + #endregion + + #region Properties + + /// + /// Gets or sets the target connection factory which will be used to create a single + /// connection. + /// + /// The target connection factory. + public IConnectionFactory TargetConnectionFactory + { + get { return targetConnectionFactory; } + set { targetConnectionFactory = value; } + } + + + /// + /// Sets the exception listener. + /// + /// The exception listener. + public IExceptionListener ExceptionListener + { + set { exceptionListener = value; } + } + + + /// + /// Gets or sets a value indicating whether the single Connection + /// should be reset (to be subsequently renewed) when a NMSException + /// is reported by the underlying Connection. + /// + /// + /// Default is false. Switch this to true + /// to automatically trigger recover based on your messaging provider's + /// exception notifications. + /// + /// Internally, this will lead to a special ExceptionListener (this + /// SingleConnectionFactory itself) being registered with the underlying + /// Connection. This can also be combined with a user-specified + /// ExceptionListener, if desired. + /// + /// + /// + /// true attempt to reconnect on exception during next access; otherwise, false. + /// + public bool ReconnectOnException + { + get { return reconnectOnException; } + set { reconnectOnException = value; } + } + + /// + /// Gets the connection monitor. + /// + /// The connection monitor. + internal object ConnectionMonitor + { + get { return connectionMonitor; } + } + + /// + /// Gets a value indicating whether this instance is started. + /// + /// + /// true if this instance is started; otherwise, false. + /// + internal bool IsStarted + { + get { return started;} + set { started = value; } + } + + /// + /// Gets the client id. + /// + /// The client id. + internal string ClientId + { + get { return clientId; } + } + + #endregion + + #region Implementation of IConnectionFactory + + /// + /// Creates the connection. + /// + /// A single shared connection + public IConnection CreateConnection() + { + lock (connectionMonitor) + { + if (connection == null) + { + InitConnection(); + } + return connection; + } + } + + /// + /// Creates the connection. + /// + /// Name of the user. + /// The password. + /// + public IConnection CreateConnection(string userName, string password) + { + throw new InvalidOperationException("SingleConnectionFactory does not support custom username and password."); + } + + public ConnectionFactory NativeConnectionFactory + { + get { return targetConnectionFactory.NativeConnectionFactory; } + } + + public object GetCertificateStore() + { + return TargetConnectionFactory.GetCertificateStore(); + } + + public string GetSSLProxyHost() + { + return TargetConnectionFactory.GetSSLProxyHost(); + } + + public string GetSSLProxyPassword() + { + return TargetConnectionFactory.GetSSLProxyPassword(); + } + + public int GetSSLProxyPort() + { + return TargetConnectionFactory.GetSSLProxyPort(); + } + + public string GetSSLProxyUser() + { + return TargetConnectionFactory.GetSSLProxyUser(); + } + + public void SetCertificateStoreType(EMSSSLStoreType type, object storeInfo) + { + TargetConnectionFactory.SetCertificateStoreType(type, storeInfo); + } + + public void SetClientID(string clientID) + { + this.clientId = clientID; + TargetConnectionFactory.SetClientID(clientID); + } + + public void SetClientTracer(StreamWriter tracer) + { + TargetConnectionFactory.SetClientTracer(tracer); + } + + public void SetConnAttemptCount(int attempts) + { + TargetConnectionFactory.SetConnAttemptCount(attempts); + } + + public void SetConnAttemptDelay(int delay) + { + TargetConnectionFactory.SetConnAttemptDelay(delay); + } + + public void SetConnAttemptTimeout(int timeout) + { + TargetConnectionFactory.SetConnAttemptTimeout(timeout); + } + + public void SetHostNameVerifier(EMSSSLHostNameVerifier verifier) + { + TargetConnectionFactory.SetHostNameVerifier(verifier); + } + + public void SetMetric(int metric) + { + TargetConnectionFactory.SetMetric(metric); + } + + public void SetMulticastDaemon(string port) + { + TargetConnectionFactory.SetMulticastDaemon(port); + } + + public void SetMulticastEnabled(bool enabled) + { + TargetConnectionFactory.SetMulticastEnabled(enabled); + } + + public void SetReconnAttemptCount(int attempts) + { + TargetConnectionFactory.SetReconnAttemptCount(attempts); + } + + public void SetReconnAttemptDelay(int delay) + { + TargetConnectionFactory.SetReconnAttemptDelay(delay); + } + + public void SetReconnAttemptTimeout(int timeout) + { + TargetConnectionFactory.SetReconnAttemptTimeout(timeout); + } + + public void SetServerUrl(string serverUrl) + { + TargetConnectionFactory.SetServerUrl(serverUrl); + } + + public void SetSSLAuthOnly(bool authOnly) + { + TargetConnectionFactory.SetSSLAuthOnly(authOnly); + } + + public void SetSSLProxy(string host, int port) + { + TargetConnectionFactory.SetSSLProxy(host, port); + } + + public void SetSSLProxyAuth(string username, string password) + { + TargetConnectionFactory.SetSSLProxyAuth(username, password); + } + + public void SetSSLTrace(bool trace) + { + TargetConnectionFactory.SetSSLTrace(trace); + } + + public void SetTargetHostName(string targetHostName) + { + TargetConnectionFactory.SetTargetHostName(targetHostName); + } + + public void SetUserName(string username) + { + TargetConnectionFactory.SetUserName(username); + } + + public void SetUserPassword(string password) + { + TargetConnectionFactory.SetUserPassword(password); + } + + public FactoryLoadBalanceMetric Metric + { + get { return TargetConnectionFactory.Metric; } + set { TargetConnectionFactory.Metric = value; } + } + + #endregion + + #region Implementation of ISerializable + + public void GetObjectData(SerializationInfo info, StreamingContext context) + { + TargetConnectionFactory.GetObjectData(info, context); + } + + #endregion + + #region Implementation of ICloneable + + public object Clone() + { + return TargetConnectionFactory.Clone(); + } + + #endregion + + /// + /// Initialize the underlying shared Connection. Closes and reinitializes the Connection if an underlying + /// Connection is present already. + /// + public void InitConnection() + { + if (TargetConnectionFactory == null) + { + throw new ArgumentException( + "'TargetConnectionFactory' is required for lazily initializing a Connection"); + } + lock (connectionMonitor) + { + if (this.target != null) + { + CloseConnection(this.target); + } + this.target = DoCreateConnection(); + PrepareConnection(this.target); + if (LOG.IsDebugEnabled) + { + LOG.Info("Established shared NMS Connection: " + this.target); + } + this.connection = GetSharedConnection(target); + } + } + + /// + /// Exception listener callback that renews the underlying single Connection. + /// + /// The exception from the messaging infrastructure. + public void OnException(EMSException exception) + { + ResetConnection(); + } + + /// + /// Prepares the connection before it is exposed. + /// The default implementation applies ExceptionListener and client id. + /// Can be overridden in subclasses. + /// + /// The Connection to prepare. + /// if thrown by any NMS API methods. + protected virtual void PrepareConnection(IConnection con) + { + if (ClientId != null) + { + con.ClientID = ClientId; + } + if (this.exceptionListener != null || this.reconnectOnException) + { + IExceptionListener listenertoUse = this.exceptionListener; + + } + /* + if (reconnectOnException) + { + //add reconnect exception handler first to exception chain. + + con.ExceptionListener += new ExceptionHandler(this.OnException); + } + if (ExceptionListener != null) + { + con.ExceptionListener += new ExceptionListener(ExceptionListener.OnException); + }*/ + } + + /// + /// Template method for obtaining a (potentially cached) Session. + /// + /// The connection to operate on. + /// The session ack mode. + /// the Session to use, or null to indicate + /// creation of a raw standard Session + public virtual ISession GetSession(IConnection con, SessionMode mode) + { + return null; + } + + + + /// + /// reate a JMS Connection via this template's ConnectionFactory. + /// + /// + protected virtual IConnection DoCreateConnection() + { + return TargetConnectionFactory.CreateConnection(); + } + + /// + /// Closes the given connection. + /// + /// The connection. + protected virtual void CloseConnection(IConnection con) + { + if (LOG.IsDebugEnabled) + { + LOG.Debug("Closing shared NMS Connection: " + this.target); + } + try + { + try + { + if (this.started) + { + this.started = false; + con.Stop(); + } + } finally + { + con.Close(); + } + } catch (Exception ex) + { + LOG.Warn("Could not close shared NMS connection.", ex); + } + } + + #region IInitializingObject Members + + /// + /// Ensure that the connection or TargetConnectionFactory are specified. + /// + public void AfterPropertiesSet() + { + if (connection == null && TargetConnectionFactory == null) + { + throw new ArgumentException("Connection or 'TargetConnectionFactory' is required."); + } + } + + #endregion + + + /// + /// Close the underlying shared connection. The provider of this ConnectionFactory needs to care for proper shutdown. + /// As this object implements an application context will automatically + /// invoke this on distruction o + /// + public void Dispose() + { + ResetConnection(); + } + + /// + /// Resets the underlying shared Connection, to be reinitialized on next access. + /// + public virtual void ResetConnection() + { + lock (connectionMonitor) + { + if (this.target != null) + { + CloseConnection(this.target); + } + this.target = null; + this.connection = null; + } + } + + /// + /// Wrap the given Connection with a proxy that delegates every method call to it + /// but suppresses close calls. This is useful for allowing application code to + /// handle a special framework Connection just like an ordinary Connection from a + /// ConnectionFactory. + /// + /// The original connection to wrap. + /// the wrapped connection + protected virtual IConnection GetSharedConnection(IConnection target) + { + lock (connectionMonitor) + { + return new CloseSupressingConnection(this, target); + } + } + } + + /// + /// Internal chained ExceptionListener for handling the internal recovery listener + /// in combination with a user-specified listener. + /// + internal class InternalChainedExceptionListener : ChainedExceptionListener + { + private IExceptionListener userListener; + + public InternalChainedExceptionListener(IExceptionListener internalListener, IExceptionListener userListener) + { + AddListener(internalListener); + if (userListener != null) + { + AddListener(userListener); + this.userListener = userListener; + } + } + + public IExceptionListener UserListener + { + get { return userListener; } + } + } + internal class CloseSupressingConnection : IConnection + { + private IConnection target; + private SingleConnectionFactory singleConnectionFactory; + + public CloseSupressingConnection(SingleConnectionFactory singleConnectionFactory, IConnection target) + { + this.target = target; + this.singleConnectionFactory = singleConnectionFactory; + } + + /// + /// Add information to show this is a shared NMS connection + /// + /// Description of connection wrapper + public override string ToString() + { + return "Shared EMS Connection: " + this.target; + } + + public string ClientID + { + get { return target.ClientID; } + set + { + // Handle set ClientID property: throw exception if not compatible. + string currentClientId = target.ClientID; + if (currentClientId != null && currentClientId.Equals(value)) + { + //ok, the values are consistent. + } + else + { + throw new IllegalStateException( + "Setting of 'ClientID' property not supported on wrapper for shared Connection since" + + "this is a shared connection that may serve any number of clients concurrently." + + "Set the 'ClientId' property on the SingleConnectionFactory instead."); + } + + } + } + + + public void Start() + { + // Handle start method: track started state. + target.Start(); + lock (singleConnectionFactory.ConnectionMonitor) + { + singleConnectionFactory.IsStarted = true; + } + } + + public void Stop() + { + //don't pass the call to the target as it would stop receiving for all clients sharing this connection. + } + + public void Close() + { + // don't pass the call to the target. + } + + public ISession CreateSession(bool transacted, int acknowledgementMode) + { + ISession session = singleConnectionFactory.GetSession(target, EmsUtils.ConvertAcknowledgementMode(acknowledgementMode)); + if (session != null) + { + return session; + } + return target.CreateSession(transacted, acknowledgementMode); + } + + public ISession CreateSession(bool transacted, SessionMode acknowledgeMode) + { + ISession session = singleConnectionFactory.GetSession(target, acknowledgeMode); + if (session != null) + { + return session; + } + return target.CreateSession(transacted, acknowledgeMode); + } + + public IExceptionListener ExceptionListener + { + get + { + IExceptionListener currentExceptionListener = target.ExceptionListener; + if (currentExceptionListener is InternalChainedExceptionListener) + { + return ((InternalChainedExceptionListener) currentExceptionListener).UserListener; + } else + { + return currentExceptionListener; + } + } + set + { + IExceptionListener currentExceptionListener = target.ExceptionListener; + if (value != null && currentExceptionListener is InternalChainedExceptionListener) + { + ((InternalChainedExceptionListener) currentExceptionListener).AddListener(value); + } else + { + throw new IllegalStateException( + "set ExceptionListener call not supported on proxy for shared Connection. " + + "Set the 'ExceptionListener' property on the SingleConnectionFactory instead. " + + "Alternatively, activate SingleConnectionFactory's 'reconnectOnException' feature, " + + "which will allow for registering further ExceptionListeners to the recovery chain."); + } + } + } + + #region Pass through implementations to the target connection + + + public event EMSExceptionHandler EMSExceptionHandler + { + add + { + target.EMSExceptionHandler += value; + } + remove + { + target.EMSExceptionHandler -= value; + } + } + + + public Connection NativeConnection + { + get { return target.NativeConnection; } + } + + public string ActiveURL + { + get { return target.ActiveURL; } + } + + public long ConnID + { + get { return target.ConnID; } + } + + public bool IsClosed + { + get { return target.IsClosed; } + } + + public bool IsSecure + { + get { return target.IsSecure; } + } + + public ConnectionMetaData MetaData + { + get { return target.MetaData; } + } + + #endregion + + + } +} \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/BrowserDelegate.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/BrowserDelegate.cs index 2219e6ab..c56cdbfc 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/BrowserDelegate.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/BrowserDelegate.cs @@ -20,6 +20,7 @@ #endregion +using Spring.Messaging.Ems.Common; using TIBCO.EMS; namespace Spring.Messaging.Ems.Core @@ -27,5 +28,5 @@ namespace Spring.Messaging.Ems.Core /// /// Delegate callback for browsing the messages in an EMS queue. /// - public delegate object BrowserDelegate(Session session, QueueBrowser browser); + public delegate object BrowserDelegate(ISession session, QueueBrowser browser); } \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/EmsGatewaySupport.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/EmsGatewaySupport.cs index 7f22bd2c..d5a7de22 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/EmsGatewaySupport.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/EmsGatewaySupport.cs @@ -20,6 +20,7 @@ using System; using Common.Logging; +using Spring.Messaging.Ems.Common; using Spring.Objects.Factory; using TIBCO.EMS; @@ -62,7 +63,7 @@ namespace Spring.Messaging.Ems.Core /// Will automatically create a EmsTemplate for the given ConnectionFactory. /// /// The connection factory. - public ConnectionFactory ConnectionFactory + public IConnectionFactory ConnectionFactory { get { @@ -83,7 +84,7 @@ namespace Spring.Messaging.Ems.Core /// /// The connection factory. /// - protected virtual EmsTemplate CreateEmsTemplate(ConnectionFactory connectionFactory) + protected virtual EmsTemplate CreateEmsTemplate(IConnectionFactory connectionFactory) { return new EmsTemplate(connectionFactory); } diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/EmsTemplate.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/EmsTemplate.cs index 41f247ca..7c839da0 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/EmsTemplate.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/EmsTemplate.cs @@ -21,6 +21,7 @@ using System; using System.Collections; using Common.Logging; +using Spring.Messaging.Ems.Common; using Spring.Messaging.Ems.Connections; using Spring.Messaging.Ems.Support; using Spring.Messaging.Ems.Support.Converter; @@ -88,9 +89,12 @@ namespace Spring.Messaging.Ems.Core private long timeToLive = Message.DEFAULT_TIME_TO_LIVE; + //TODO make optimizations later based on TIBCO EMS having thread safe sessions + /* private EmsResources emsResources = new EmsResources(); private bool cacheEmsResources = true; + */ #endregion @@ -113,7 +117,7 @@ namespace Spring.Messaging.Ems.Core /// Create a new EmsTemplate, given a ConnectionFactory. /// the ConnectionFactory to obtain Connections from /// - public EmsTemplate(ConnectionFactory connectionFactory) + public EmsTemplate(IConnectionFactory connectionFactory) : this() { ConnectionFactory = connectionFactory; @@ -192,27 +196,27 @@ namespace Spring.Messaging.Ems.Core { AssertUtils.ArgumentNotNull(action, "Callback object must not be null"); - Connection con = null; - Session session = null; - bool sessionInTLS = true; + IConnection conToClose = null; + ISession sessionToClose = null; + // bool sessionInTLS = true; //NOTE: Not closing session or connection unless session is not returned from // ConnectionFactoryUtils.DoGetTransactionalSession and CacheEmsResources is set to false try { - Session sessionToUse = + ISession sessionToUse = ConnectionFactoryUtils.DoGetTransactionalSession(ConnectionFactory, transactionalResourceFactory, startConnection); if (sessionToUse == null) { - sessionInTLS = false; - con = CreateConnection(); - session = CreateSession(con); + //sessionInTLS = false; + conToClose = CreateConnection(); + sessionToClose = CreateSession(conToClose); if (startConnection) { - con.Start(); + conToClose.Start(); } - sessionToUse = session; + sessionToUse = sessionToClose; } if (logger.IsDebugEnabled) { @@ -222,11 +226,14 @@ namespace Spring.Messaging.Ems.Core } finally { + EmsUtils.CloseSession(sessionToClose); + ConnectionFactoryUtils.ReleaseConnection(conToClose, ConnectionFactory, startConnection); + /* if (!sessionInTLS && !CacheEmsResources) { EmsUtils.CloseSession(session); ConnectionFactoryUtils.ReleaseConnection(con, ConnectionFactory, startConnection); - } + }*/ } } @@ -387,6 +394,7 @@ namespace Spring.Messaging.Ems.Core set { timeToLive = value; } } + /// /// Gets or sets a value indicating whether the EmsTemplate should itself /// be responsible for caching EMS Connection/Session/MessageProducer as compared to @@ -396,11 +404,11 @@ namespace Spring.Messaging.Ems.Core /// /// Connection/Session/MessageProducer are thread-safe classes in TIBCO EMS. /// true to locally cache ems resources; otherwise, false. - virtual public bool CacheEmsResources +/* virtual public bool CacheEmsResources { get { return cacheEmsResources; } set { cacheEmsResources = value; } - } + }*/ #endregion @@ -428,7 +436,7 @@ namespace Spring.Messaging.Ems.Core /// an appropriate Connection fetched from the holder, /// or null if none found /// - protected virtual Connection GetConnection(EmsResourceHolder holder) + protected virtual IConnection GetConnection(EmsResourceHolder holder) { return holder.GetConnection(); } @@ -440,7 +448,7 @@ namespace Spring.Messaging.Ems.Core /// an appropriate Session fetched from the holder, /// or null if none found /// - protected virtual Session GetSession(EmsResourceHolder holder) + protected virtual ISession GetSession(EmsResourceHolder holder) { return holder.GetSession(); } @@ -463,9 +471,9 @@ namespace Spring.Messaging.Ems.Core /// /// /// - protected virtual MessageProducer CreateProducer(Session session, Destination destination) + protected virtual IMessageProducer CreateProducer(ISession session, Destination destination) { - MessageProducer producer = DoCreateProducer(session, destination); + IMessageProducer producer = DoCreateProducer(session, destination); if (!MessageIdEnabled) { producer.DisableMessageID = true; @@ -492,7 +500,7 @@ namespace Spring.Messaging.Ems.Core /// /// true if the session is locally transacted; otherwise, false. /// - protected virtual bool IsSessionLocallyTransacted(Session session) + protected virtual bool IsSessionLocallyTransacted(ISession session) { return SessionTransacted && !ConnectionFactoryUtils.IsSessionTransactional(session, ConnectionFactory); @@ -511,8 +519,10 @@ namespace Spring.Messaging.Ems.Core /// the new EMS MessageProducer /// /// If there is any problem accessing the EMS API - protected virtual MessageProducer DoCreateProducer(Session session, Destination destination) + protected virtual IMessageProducer DoCreateProducer(ISession session, Destination destination) { + return session.CreateProducer(destination); + /* if (CacheEmsResources) { if (destination == null) @@ -523,7 +533,7 @@ namespace Spring.Messaging.Ems.Core } return emsResources.UnspecifiedDestinationMessageProducer; } - MessageProducer producer = (MessageProducer)emsResources.Producers[destination]; + IMessageProducer producer = (IMessageProducer)emsResources.Producers[destination]; if (producer != null) { #region Logging @@ -553,7 +563,7 @@ namespace Spring.Messaging.Ems.Core else { return session.CreateProducer(destination); - } + }*/ } /// Create a EMS MessageConsumer for the given Session and Destination. @@ -567,8 +577,8 @@ namespace Spring.Messaging.Ems.Core /// the new EMS MessageConsumer /// /// If there is any problem accessing the EMS API - protected virtual MessageConsumer CreateConsumer(Session session, Destination destination, - string messageSelector) + protected virtual IMessageConsumer CreateConsumer(ISession session, Destination destination, + string messageSelector) { // Only pass in the NoLocal flag in case of a Topic: // Some EMS providers, such as WebSphere MQ 6.0, throw IllegalStateException @@ -592,7 +602,7 @@ namespace Spring.Messaging.Ems.Core /// A EMS Connection /// /// If there is any problem accessing the EMS API - protected override Connection CreateConnection() +/* protected override IConnection CreateConnection() { if (CacheEmsResources) { @@ -607,7 +617,7 @@ namespace Spring.Messaging.Ems.Core { return ConnectionFactory.CreateConnection(); } - } + }*/ /// Create a EMS Session for the given Connection. /// @@ -620,7 +630,7 @@ namespace Spring.Messaging.Ems.Core /// the new EMS Session /// /// If there is any problem accessing the EMS API - protected override Session CreateSession(Connection con) +/* protected override ISession CreateSession(IConnection con) { if (CacheEmsResources) { @@ -634,7 +644,7 @@ namespace Spring.Messaging.Ems.Core { return con.CreateSession(SessionTransacted, SessionAcknowledgeMode); } - } + }*/ /// /// Send the given message. @@ -642,7 +652,7 @@ namespace Spring.Messaging.Ems.Core /// The session to operate on. /// The destination to send to. /// The message creator delegate callback to create a Message. - protected internal virtual void DoSend(Session session, Destination destination, MessageCreatorDelegate messageCreatorDelegate) + protected internal virtual void DoSend(ISession session, Destination destination, MessageCreatorDelegate messageCreatorDelegate) { AssertUtils.ArgumentNotNull(messageCreatorDelegate, "IMessageCreatorDelegate must not be null"); DoSend(session, destination, null, messageCreatorDelegate); @@ -654,7 +664,7 @@ namespace Spring.Messaging.Ems.Core /// The session to operate on. /// The destination to send to. /// The message creator callback to create a Message. - protected internal virtual void DoSend(Session session, Destination destination, IMessageCreator messageCreator) + protected internal virtual void DoSend(ISession session, Destination destination, IMessageCreator messageCreator) { AssertUtils.ArgumentNotNull(messageCreator, "IMessageCreator must not be null"); DoSend(session, destination, messageCreator, null); @@ -670,12 +680,12 @@ namespace Spring.Messaging.Ems.Core /// delegate callback to create a EMS Message /// /// If there is any problem accessing the EMS API - protected internal virtual void DoSend(Session session, Destination destination, IMessageCreator messageCreator, + protected internal virtual void DoSend(ISession session, Destination destination, IMessageCreator messageCreator, MessageCreatorDelegate messageCreatorDelegate) { - MessageProducer producer = CreateProducer(session, destination); + IMessageProducer producer = CreateProducer(session, destination); try { @@ -713,7 +723,7 @@ namespace Spring.Messaging.Ems.Core /// the EMS Message to send /// /// If there is any problem accessing the EMS API - protected virtual void DoSend(MessageProducer producer, Message message) + protected virtual void DoSend(IMessageProducer producer, Message message) { if (ExplicitQosEnabled) { @@ -1171,7 +1181,7 @@ namespace Spring.Messaging.Ems.Core /// The destination to receive from. /// The message selector for this consumer (can be null /// The Message received, or null if none. - protected virtual Message DoReceive(Session session, Destination destination, string messageSelector) + protected virtual Message DoReceive(ISession session, Destination destination, string messageSelector) { return DoReceive(session, CreateConsumer(session, destination, messageSelector)); } @@ -1182,7 +1192,7 @@ namespace Spring.Messaging.Ems.Core /// The session to operate on. /// The consumer to receive with. /// The Message received, or null if none - protected virtual Message DoReceive(Session session, MessageConsumer consumer) + protected virtual Message DoReceive(ISession session, IMessageConsumer consumer) { try { @@ -1507,7 +1517,7 @@ namespace Spring.Messaging.Ems.Core public object BrowseSelectedWithDelegate(Queue queue, string messageSelector, BrowserDelegate action) { AssertUtils.ArgumentNotNull(action, "action"); - return Execute(delegate(Session session) + return Execute(delegate(ISession session) { QueueBrowser browser = CreateBrowser(session, queue, messageSelector); try @@ -1534,7 +1544,7 @@ namespace Spring.Messaging.Ems.Core public object BrowseSelectedWithDelegate(string queueName, string messageSelector, BrowserDelegate action) { AssertUtils.ArgumentNotNull(action, "action"); - return Execute(delegate(Session session) + return Execute(delegate(ISession session) { Queue queue = (Queue)DestinationResolver.ResolveDestinationName(session, queueName, false); QueueBrowser browser = CreateBrowser(session, queue, messageSelector); @@ -1558,7 +1568,7 @@ namespace Spring.Messaging.Ems.Core /// The queue. /// The selector. /// A new queue browser - protected virtual QueueBrowser CreateBrowser(Session session, Queue queue, string selector) + protected virtual QueueBrowser CreateBrowser(ISession session, Queue queue, string selector) { return session.CreateBrowser(queue, selector); } @@ -1588,22 +1598,22 @@ namespace Spring.Messaging.Ems.Core get { return enclosingTemplateInstance; } } - public virtual Connection GetConnection(EmsResourceHolder holder) + public virtual IConnection GetConnection(EmsResourceHolder holder) { return EnclosingInstance.GetConnection(holder); } - public virtual Session GetSession(EmsResourceHolder holder) + public virtual ISession GetSession(EmsResourceHolder holder) { return EnclosingInstance.GetSession(holder); } - public virtual Connection CreateConnection() + public virtual IConnection CreateConnection() { return EnclosingInstance.CreateConnection(); } - public virtual Session CreateSession(Connection con) + public virtual ISession CreateSession(IConnection con) { return EnclosingInstance.CreateSession(con); } @@ -1633,9 +1643,9 @@ namespace Spring.Messaging.Ems.Core } - public object DoInEms(Session session) + public object DoInEms(ISession session) { - MessageProducer producer = jmsTemplate.CreateProducer(session, null); + IMessageProducer producer = jmsTemplate.CreateProducer(session, null); try { if (producerCallback != null) @@ -1674,7 +1684,7 @@ namespace Spring.Messaging.Ems.Core this.destination = destination; } - public object DoInEms(Session session) + public object DoInEms(ISession session) { if (destination != null) { @@ -1711,7 +1721,7 @@ namespace Spring.Messaging.Ems.Core this.messagePostProcessorDelegate = messagePostProcessorDelegate; } - public Message CreateMessage(Session session) + public Message CreateMessage(ISession session) { Message msg = jmsTemplate.MessageConverter.ToMessage(objectToConvert, session); if (messagePostProcessor != null) @@ -1749,7 +1759,7 @@ namespace Spring.Messaging.Ems.Core this.messageSelector = messageSelector; } - public object DoInEms(Session session) + public object DoInEms(ISession session) { if (destination != null) { @@ -1774,7 +1784,7 @@ namespace Spring.Messaging.Ems.Core this.del = del; } - public object DoInEms(Session session) + public object DoInEms(ISession session) { return del(session); } @@ -1787,27 +1797,27 @@ namespace Spring.Messaging.Ems.Core /// This is a TIBCO specific class so that we can reuse connections, session, and /// message producers instead of creating/destroying them on each operation. /// - internal class EmsResources +/* internal class EmsResources { - private Connection connection; - private Session session; + private IConnection connection; + private ISession session; private IDictionary cachedProducers = new Hashtable(); - private MessageProducer cachedUnspecifiedDestinationMessageProducer; + private IMessageProducer cachedUnspecifiedDestinationMessageProducer; - public Connection Connection + public IConnection Connection { get { return connection; } set { connection = value; } } - public Session Session + public ISession Session { get { return session; } set { session = value; } } - public MessageProducer UnspecifiedDestinationMessageProducer + public IMessageProducer UnspecifiedDestinationMessageProducer { get { return cachedUnspecifiedDestinationMessageProducer; } set { cachedUnspecifiedDestinationMessageProducer = value; } @@ -1819,7 +1829,7 @@ namespace Spring.Messaging.Ems.Core get { return cachedProducers; } set { cachedProducers = value; } } - } + }*/ internal class SimpleMessageCreator : IMessageCreator @@ -1833,7 +1843,7 @@ namespace Spring.Messaging.Ems.Core this.objectToConvert = objectToConvert; } - public Message CreateMessage(Session session) + public Message CreateMessage(ISession session) { return jmsTemplate.MessageConverter.ToMessage(objectToConvert, session); } @@ -1880,7 +1890,7 @@ namespace Spring.Messaging.Ems.Core } - public object DoInEms(Session session) + public object DoInEms(ISession session) { if (destination == null) { diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IBrowserCallback.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IBrowserCallback.cs index 7f7b44b2..c55f7bef 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IBrowserCallback.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IBrowserCallback.cs @@ -20,6 +20,7 @@ #endregion +using Spring.Messaging.Ems.Common; using TIBCO.EMS; namespace Spring.Messaging.Ems.Core @@ -42,6 +43,6 @@ namespace Spring.Messaging.Ems.Core /// The browser. /// The object from working with the Session and QueueBrowser, may be null /// If there is any problem when accessing EMS API - object DoInEms(Session session, QueueBrowser browser); + object DoInEms(ISession session, QueueBrowser browser); } } \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IEmsOperations.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IEmsOperations.cs index d4a23a45..969586fd 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IEmsOperations.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IEmsOperations.cs @@ -18,7 +18,6 @@ #endregion -using Spring.Messaging.Ems.Core; using TIBCO.EMS; namespace Spring.Messaging.Ems.Core diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IMessageCreator.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IMessageCreator.cs index 620fc10b..8a95905b 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IMessageCreator.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IMessageCreator.cs @@ -18,6 +18,7 @@ #endregion +using Spring.Messaging.Ems.Common; using TIBCO.EMS; namespace Spring.Messaging.Ems.Core @@ -37,6 +38,6 @@ namespace Spring.Messaging.Ems.Core /// the Message to be sent /// /// EMSException if thrown by EMS API methods - Message CreateMessage(Session session); + Message CreateMessage(ISession session); } } diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IProducerCallback.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IProducerCallback.cs index 4bb97203..c5334682 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IProducerCallback.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/IProducerCallback.cs @@ -18,7 +18,7 @@ #endregion -using TIBCO.EMS; +using Spring.Messaging.Ems.Common; namespace Spring.Messaging.Ems.Core { @@ -42,7 +42,7 @@ namespace Spring.Messaging.Ems.Core /// /// a result object from working with the Session, if any (can be null) /// - object DoInEms(Session session, MessageProducer producer); + object DoInEms(ISession session, IMessageProducer producer); } } \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/ISessionCallback.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/ISessionCallback.cs index 53ff5678..635eb094 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/ISessionCallback.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/ISessionCallback.cs @@ -18,7 +18,7 @@ #endregion -using TIBCO.EMS; +using Spring.Messaging.Ems.Common; namespace Spring.Messaging.Ems.Core { @@ -42,6 +42,6 @@ namespace Spring.Messaging.Ems.Core /// a result object from working with the Session, if any (so can be null) /// /// EMSException if there is any problem - object DoInEms(Session session); + object DoInEms(ISession session); } } diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/MessageCreatorDelegate.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/MessageCreatorDelegate.cs index fd12d227..cf1d432a 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/MessageCreatorDelegate.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/MessageCreatorDelegate.cs @@ -18,6 +18,7 @@ #endregion +using Spring.Messaging.Ems.Common; using TIBCO.EMS; namespace Spring.Messaging.Ems.Core @@ -31,5 +32,5 @@ namespace Spring.Messaging.Ems.Core /// the Message to be sent /// /// EMSException if thrown by EMS API methods - public delegate Message MessageCreatorDelegate(Session session); + public delegate Message MessageCreatorDelegate(ISession session); } diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/ProducerDelegate.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/ProducerDelegate.cs index d7bf6cd7..b61d95f2 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/ProducerDelegate.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/ProducerDelegate.cs @@ -20,7 +20,7 @@ #endregion -using TIBCO.EMS; +using Spring.Messaging.Ems.Common; namespace Spring.Messaging.Ems.Core { @@ -33,6 +33,6 @@ namespace Spring.Messaging.Ems.Core /// /// a result object from working with the Session, if any (can be null) /// - public delegate object ProducerDelegate(Session session, MessageProducer producer); + public delegate object ProducerDelegate(ISession session, IMessageProducer producer); } \ No newline at end of file diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/SessionDelegate.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/SessionDelegate.cs index c79cc6fd..c9536add 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/SessionDelegate.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Core/SessionDelegate.cs @@ -18,7 +18,7 @@ #endregion -using TIBCO.EMS; +using Spring.Messaging.Ems.Common; namespace Spring.Messaging.Ems.Core { @@ -35,5 +35,5 @@ namespace Spring.Messaging.Ems.Core /// /// EMSException if there is any problem /// Mark Pollack - public delegate object SessionDelegate(Session session); + public delegate object SessionDelegate(ISession session); } diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/AbstractListenerContainer.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/AbstractListenerContainer.cs index f0139757..cff45414 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/AbstractListenerContainer.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/AbstractListenerContainer.cs @@ -21,6 +21,7 @@ using System; using Common.Logging; using Spring.Context; +using Spring.Messaging.Ems.Common; using Spring.Messaging.Ems.Connections; using Spring.Messaging.Ems.Support; using Spring.Messaging.Ems.Support.Destinations; @@ -61,7 +62,7 @@ namespace Spring.Messaging.Ems.Listener private string objectName; - private Connection sharedConnection; + private IConnection sharedConnection; private bool sharedConnectionStarted = false; @@ -193,7 +194,7 @@ namespace Spring.Messaging.Ems.Listener /// shared Connection, or if the Connection hasn't been initialized yet. /// /// - protected Connection SharedConnection + protected IConnection SharedConnection { get { @@ -446,9 +447,9 @@ namespace Spring.Messaging.Ems.Listener /// /// the prepared Connection /// if the creation failed. - protected virtual Connection CreateSharedConnection() + protected virtual IConnection CreateSharedConnection() { - Connection con = CreateConnection(); + IConnection con = CreateConnection(); try { PrepareSharedConnection(con); @@ -470,7 +471,7 @@ namespace Spring.Messaging.Ems.Listener /// /// The connection to prepare. /// If the preparation efforts failed. - protected virtual void PrepareSharedConnection(Connection connection) + protected virtual void PrepareSharedConnection(IConnection connection) { if (ClientId != null) { diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/AbstractMessageListenerContainer.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/AbstractMessageListenerContainer.cs index 26c13ed1..bb519da9 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/AbstractMessageListenerContainer.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/AbstractMessageListenerContainer.cs @@ -20,6 +20,7 @@ using System; using Common.Logging; +using Spring.Messaging.Ems.Common; using Spring.Messaging.Ems.Core; using Spring.Messaging.Ems.Support; using Spring.Util; @@ -296,7 +297,7 @@ namespace Spring.Messaging.Ems.Listener /// /// /// - public virtual void ExecuteListener(Session session, Message message) + public virtual void ExecuteListener(ISession session, Message message) { try { @@ -318,7 +319,7 @@ namespace Spring.Messaging.Ems.Listener /// /// /// - protected virtual void DoExecuteListener(Session session, Message message) + protected virtual void DoExecuteListener(ISession session, Message message) { if (!AcceptMessagesWhileStopping && !IsRunning) { @@ -353,7 +354,7 @@ namespace Spring.Messaging.Ems.Listener /// The received message. /// If thrown by EMS API methods. /// - protected virtual void InvokeListener(Session session, Message message) + protected virtual void InvokeListener(ISession session, Message message) { object listener = MessageListener; if (listener is ISessionAwareMessageListener) @@ -386,13 +387,13 @@ namespace Spring.Messaging.Ems.Listener /// If thrown by EMS API methods. /// /// - protected virtual void DoInvokeListener(ISessionAwareMessageListener listener, Session session, Message message) + protected virtual void DoInvokeListener(ISessionAwareMessageListener listener, ISession session, Message message) { - Connection conToClose = null; - Session sessionToClose = null; + IConnection conToClose = null; + ISession sessionToClose = null; try { - Session sessionToUse = session; + ISession sessionToUse = session; if (!ExposeListenerSession) { //We need to expose a separate Session. @@ -442,7 +443,7 @@ namespace Spring.Messaging.Ems.Listener /// The session to commit. /// The message to acknowledge. /// In case of commit failure - protected virtual void CommitIfNecessary(Session session, Message message) + protected virtual void CommitIfNecessary(ISession session, Message message) { // Commit session or acknowledge message if (session.Transacted) @@ -474,7 +475,7 @@ namespace Spring.Messaging.Ems.Listener /// true if the is session locally transacted; otherwise, false. /// /// - protected virtual bool IsSessionLocallyTransacted(Session session) + protected virtual bool IsSessionLocallyTransacted(ISession session) { return SessionTransacted; } @@ -485,7 +486,7 @@ namespace Spring.Messaging.Ems.Listener /// /// The session to rollback. /// In case of a rollback error - protected virtual void RollbackIfNecessary(Session session) + protected virtual void RollbackIfNecessary(ISession session) { if (session.Transacted && IsSessionLocallyTransacted(session)) { @@ -499,7 +500,7 @@ namespace Spring.Messaging.Ems.Listener /// The session to rollback. /// The thrown application exception. /// in case of a rollback error. - protected virtual void RollbackOnExceptionIfNecessary(Session session, Exception ex) + protected virtual void RollbackOnExceptionIfNecessary(ISession session, Exception ex) { try { diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/Adapter/MessageListenerAdapter.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/Adapter/MessageListenerAdapter.cs index c10489e8..8d38cda3 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/Adapter/MessageListenerAdapter.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/Adapter/MessageListenerAdapter.cs @@ -3,6 +3,7 @@ using System.Collections; using System.Reflection; using Common.Logging; using Spring.Expressions; +using Spring.Messaging.Ems.Common; using Spring.Messaging.Ems.Support; using Spring.Messaging.Ems.Support.Converter; using Spring.Messaging.Ems.Support.Destinations; @@ -248,7 +249,7 @@ namespace Spring.Messaging.Ems.Listener.Adapter /// /// The incoming message. /// The session to operate on. - public void OnMessage(Message message, Session session) + public void OnMessage(Message message, ISession session) { if (handlerObject != this) { @@ -380,7 +381,7 @@ namespace Spring.Messaging.Ems.Listener.Adapter /// The result object to handle (never null). /// The original request message. /// The session to operate on (may be null). - protected virtual void HandleResult(object result, Message request, Session session) + protected virtual void HandleResult(object result, Message request, ISession session) { if (session != null) { @@ -412,7 +413,7 @@ namespace Spring.Messaging.Ems.Listener.Adapter /// the JMS Message (never null) /// If there was an error in message conversion /// if thrown by EMS API methods - protected virtual Message BuildMessage(Session session, Object result) + protected virtual Message BuildMessage(ISession session, Object result) { IMessageConverter converter = MessageConverter; if (converter != null) @@ -461,7 +462,7 @@ namespace Spring.Messaging.Ems.Listener.Adapter /// the response destination (never null) /// if thrown by EMS API methods /// if no destination can be determined. - protected virtual Destination GetResponseDestination(Message request, Message response, Session session) + protected virtual Destination GetResponseDestination(Message request, Message response, ISession session) { Destination replyTo = request.ReplyTo; if (replyTo == null) @@ -482,7 +483,7 @@ namespace Spring.Messaging.Ems.Listener.Adapter /// /// The session to operate on. /// The located destination - protected virtual Destination ResolveDefaultResponseDestination(Session session) + protected virtual Destination ResolveDefaultResponseDestination(ISession session) { Destination dest = defaultResponseDestination as Destination; if (dest != null) @@ -505,9 +506,9 @@ namespace Spring.Messaging.Ems.Listener.Adapter /// The session to operate on. /// The destination to send to. /// The outgoing message about to be sent. - protected virtual void SendResponse(Session session, Destination destination, Message response) + protected virtual void SendResponse(ISession session, Destination destination, Message response) { - MessageProducer producer = session.CreateProducer(destination); + IMessageProducer producer = session.CreateProducer(destination); try { PostProcessProducer(producer, response); @@ -525,7 +526,7 @@ namespace Spring.Messaging.Ems.Listener.Adapter /// /// The producer that will be used to send the message. /// The outgoing message about to be sent. - protected virtual void PostProcessProducer(MessageProducer producer, Message response) + protected virtual void PostProcessProducer(IMessageProducer producer, Message response) { } diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/ISessionAwareMessageListener.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/ISessionAwareMessageListener.cs index c59f311f..9572b226 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/ISessionAwareMessageListener.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/ISessionAwareMessageListener.cs @@ -18,6 +18,7 @@ #endregion +using Spring.Messaging.Ems.Common; using TIBCO.EMS; namespace Spring.Messaging.Ems.Listener @@ -46,6 +47,6 @@ namespace Spring.Messaging.Ems.Listener /// the underlying EMS Session /// /// EMSException if thrown by EMS methods - void OnMessage(Message message, Session session); + void OnMessage(Message message, ISession session); } } diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/LocallyExposedEmsResourceHolder.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/LocallyExposedEmsResourceHolder.cs index 4a13e426..c8906c57 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/LocallyExposedEmsResourceHolder.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/LocallyExposedEmsResourceHolder.cs @@ -18,8 +18,8 @@ #endregion +using Spring.Messaging.Ems.Common; using Spring.Messaging.Ems.Connections; -using TIBCO.EMS; namespace Spring.Messaging.Ems.Listener { @@ -35,7 +35,7 @@ namespace Spring.Messaging.Ems.Listener /// Initializes a new instance of the class. /// /// The session. - public LocallyExposedEmsResourceHolder(Session session) : base(session) + public LocallyExposedEmsResourceHolder(ISession session) : base(session) { } diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/SimpleMessageListenerContainer.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/SimpleMessageListenerContainer.cs index f58cac38..f27a7745 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/SimpleMessageListenerContainer.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Listener/SimpleMessageListenerContainer.cs @@ -22,6 +22,7 @@ using System; using System.Threading; using Common.Logging; using Spring.Collections; +using Spring.Messaging.Ems.Common; using Spring.Messaging.Ems.Support; using Spring.Transaction.Support; using Spring.Util; @@ -175,7 +176,7 @@ namespace Spring.Messaging.Ems.Listener /// Registers this listener container as EMS ExceptionListener on the shared connection. /// /// - protected override void PrepareSharedConnection(Connection connection) + protected override void PrepareSharedConnection(IConnection connection) { base.PrepareSharedConnection(connection); connection.ExceptionListener = this; @@ -282,11 +283,11 @@ namespace Spring.Messaging.Ems.Listener { this.sessions = new HashedSet(); this.consumers = new HashedSet(); - Connection con = SharedConnection; + IConnection con = SharedConnection; for (int i = 0; i < this.concurrentConsumers; i++) { - Session session = CreateSession(SharedConnection); - MessageConsumer consumer = CreateListenerConsumer(session); + ISession session = CreateSession(SharedConnection); + IMessageConsumer consumer = CreateListenerConsumer(session); this.sessions.Add(session); this.consumers.Add(consumer); } @@ -301,14 +302,14 @@ namespace Spring.Messaging.Ems.Listener /// The session to work on. /// the MessageConsumer"/> /// if thrown by EMS methods - private MessageConsumer CreateListenerConsumer(Session session) + private IMessageConsumer CreateListenerConsumer(ISession session) { Destination destination = Destination; if (destination == null) { destination = ResolveDestinationName(session, DestinationName); } - MessageConsumer consumer = CreateConsumer(session, destination); + IMessageConsumer consumer = CreateConsumer(session, destination); consumer.MessageListener = new SimpleMessageListener(this, session); return consumer; @@ -325,7 +326,7 @@ namespace Spring.Messaging.Ems.Listener if (consumers != null) { logger.Debug("Closing NMS MessageConsumers"); - foreach (MessageConsumer messageConsumer in consumers) + foreach (IMessageConsumer messageConsumer in consumers) { EmsUtils.CloseMessageConsumer(messageConsumer); } @@ -333,7 +334,7 @@ namespace Spring.Messaging.Ems.Listener if (sessions != null) { logger.Debug("Closing NMS Sessions"); - foreach (Session session in sessions) + foreach (ISession session in sessions) { EmsUtils.CloseSession(session); } @@ -350,7 +351,7 @@ namespace Spring.Messaging.Ems.Listener /// The session to create a MessageConsumer for. /// The destination to create a MessageConsumer for. /// The new MessageConsumer - protected MessageConsumer CreateConsumer(Session session, Destination destination) + protected IMessageConsumer CreateConsumer(ISession session, Destination destination) { // Only pass in the NoLocal flag in case of a Topic: // Some EMS providers, such as WebSphere MQ 6.0, throw IllegalStateException @@ -377,9 +378,9 @@ namespace Spring.Messaging.Ems.Listener internal class SimpleMessageListener : IMessageListener { private SimpleMessageListenerContainer container; - private Session session; + private ISession session; - public SimpleMessageListener(SimpleMessageListenerContainer container, Session session) + public SimpleMessageListener(SimpleMessageListenerContainer container, ISession session) { this.container = container; this.session = session; diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/IMessageConverter.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/IMessageConverter.cs index 7dc82fac..d9854a58 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/IMessageConverter.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/IMessageConverter.cs @@ -18,6 +18,7 @@ #endregion +using Spring.Messaging.Ems.Common; using TIBCO.EMS; namespace Spring.Messaging.Ems.Support.Converter @@ -42,7 +43,7 @@ namespace Spring.Messaging.Ems.Support.Converter /// /// EMSException if thrown by EMS API methods /// MessageConversionException in case of conversion failure - Message ToMessage(object objectToConvert, Session session); + Message ToMessage(object objectToConvert, ISession session); /// Convert from a EMS Message to a .NET object. /// the message to convert diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/SimpleMessageConverter.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/SimpleMessageConverter.cs index a05c7eb2..31424d6e 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/SimpleMessageConverter.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/SimpleMessageConverter.cs @@ -20,6 +20,7 @@ using System; using System.Collections; +using Spring.Messaging.Ems.Common; using TIBCO.EMS; namespace Spring.Messaging.Ems.Support.Converter @@ -49,7 +50,7 @@ namespace Spring.Messaging.Ems.Support.Converter /// /// EMSException if thrown by EMS API methods /// MessageConversionException in case of conversion failure - public Message ToMessage(object objectToConvert, Session session) + public Message ToMessage(object objectToConvert, ISession session) { if (objectToConvert is Message) { @@ -111,7 +112,7 @@ namespace Spring.Messaging.Ems.Support.Converter /// the resulting message /// /// EMSException if thrown by EMS methods - protected virtual TextMessage CreateMessageForString(string text, Session session) + protected virtual TextMessage CreateMessageForString(string text, ISession session) { return session.CreateTextMessage((text)); } @@ -124,7 +125,7 @@ namespace Spring.Messaging.Ems.Support.Converter /// the resulting message /// /// EMSException if thrown by EMS methods - protected virtual BytesMessage CreateMessageForByteArray(byte[] bytes, Session session) + protected virtual BytesMessage CreateMessageForByteArray(byte[] bytes, ISession session) { BytesMessage message = session.CreateBytesMessage(); message.WriteBytes(bytes); @@ -139,7 +140,7 @@ namespace Spring.Messaging.Ems.Support.Converter /// the resulting message /// /// EMSException if thrown by EMS methods - protected virtual MapMessage CreateMessageForMap(IDictionary map, Session session) + protected virtual MapMessage CreateMessageForMap(IDictionary map, ISession session) { MapMessage mapMessage = session.CreateMapMessage(); foreach (DictionaryEntry entry in map) @@ -164,7 +165,7 @@ namespace Spring.Messaging.Ems.Support.Converter /// /// EMSException if thrown by EMS methods protected virtual ObjectMessage CreateMessageForSerializable( - object objectToSend, Session session) + object objectToSend, ISession session) { return session.CreateObjectMessage(objectToSend); } diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/XmlMessageConverter.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/XmlMessageConverter.cs index 07fea792..9ec7cb3d 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/XmlMessageConverter.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Converter/XmlMessageConverter.cs @@ -4,6 +4,7 @@ using System.IO; using System.Text; using System.Xml; using System.Xml.Serialization; +using Spring.Messaging.Ems.Common; using TIBCO.EMS; namespace Spring.Messaging.Ems.Support.Converter @@ -38,7 +39,7 @@ namespace Spring.Messaging.Ems.Support.Converter /// the NMS Message /// NMSException if thrown by NMS API methods /// MessageConversionException in case of conversion failure - public Message ToMessage(object objectToConvert, Session session) + public Message ToMessage(object objectToConvert, ISession session) { if (objectToConvert == null) { diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/DynamicDestinationResolver.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/DynamicDestinationResolver.cs index e70c3394..11d2a1c4 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/DynamicDestinationResolver.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/DynamicDestinationResolver.cs @@ -18,6 +18,7 @@ #endregion +using Spring.Messaging.Ems.Common; using Spring.Util; using TIBCO.EMS; @@ -42,7 +43,7 @@ namespace Spring.Messaging.Ems.Support.Destinations /// the EMS destination (either a topic or a queue) /// /// EMSException if resolution failed - public Destination ResolveDestinationName(Session session, string destinationName, bool pubSubDomain) + public Destination ResolveDestinationName(ISession session, string destinationName, bool pubSubDomain) { AssertUtils.ArgumentNotNull(session, "Session must not be null"); AssertUtils.ArgumentNotNull(destinationName, "Destination name must not be null"); @@ -50,10 +51,7 @@ namespace Spring.Messaging.Ems.Support.Destinations { return ResolveTopic(session, destinationName); } - else - { - return ResolveQueue(session, destinationName); - } + return ResolveQueue(session, destinationName); } @@ -65,7 +63,7 @@ namespace Spring.Messaging.Ems.Support.Destinations /// the EMS Topic name /// /// EMSException if resolution failed - protected internal virtual Destination ResolveTopic(Session session, System.String topicName) + protected internal virtual Destination ResolveTopic(ISession session, string topicName) { return session.CreateTopic(topicName); } @@ -78,7 +76,7 @@ namespace Spring.Messaging.Ems.Support.Destinations /// the EMS Queue name /// /// EMSException if resolution failed - protected internal virtual Destination ResolveQueue(Session session, string queueName) + protected internal virtual Destination ResolveQueue(ISession session, string queueName) { return session.CreateQueue(queueName); } diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/EmsDestinationAccessor.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/EmsDestinationAccessor.cs index 45500292..16d40db5 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/EmsDestinationAccessor.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/EmsDestinationAccessor.cs @@ -18,6 +18,7 @@ #endregion +using Spring.Messaging.Ems.Common; using Spring.Util; using TIBCO.EMS; @@ -102,7 +103,7 @@ namespace Spring.Messaging.Ems.Support.Destinations /// Name of the destination. /// The located Destination /// If resolution failed. - public virtual Destination ResolveDestinationName(Session session, System.String destinationName) + public virtual Destination ResolveDestinationName(ISession session, System.String destinationName) { return DestinationResolver.ResolveDestinationName(session, destinationName, PubSubDomain); } diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/IDestinationResolver.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/IDestinationResolver.cs index f3b94207..92a7f45e 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/IDestinationResolver.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/Destinations/IDestinationResolver.cs @@ -18,6 +18,7 @@ #endregion +using Spring.Messaging.Ems.Common; using TIBCO.EMS; namespace Spring.Messaging.Ems.Support.Destinations @@ -52,7 +53,7 @@ namespace Spring.Messaging.Ems.Support.Destinations /// the EMS destination (either a topic or a queue) /// /// EMSException if resolution failed - Destination ResolveDestinationName(Session session, string destinationName, bool pubSubDomain); + Destination ResolveDestinationName(ISession session, string destinationName, bool pubSubDomain); } } diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsAccessor.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsAccessor.cs index d80c5824..cc17827f 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsAccessor.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsAccessor.cs @@ -20,6 +20,7 @@ using System; using Common.Logging; +using Spring.Messaging.Ems.Common; using Spring.Objects.Factory; using TIBCO.EMS; @@ -44,7 +45,7 @@ namespace Spring.Messaging.Ems.Support #region Fields - private ConnectionFactory connectionFactory; + private IConnectionFactory connectionFactory; private bool sessionTransacted = false; @@ -59,7 +60,7 @@ namespace Spring.Messaging.Ems.Support /// Gets or sets the connection factory to use for obtaining EMS Connections. /// /// The connection factory. - virtual public ConnectionFactory ConnectionFactory + virtual public IConnectionFactory ConnectionFactory { get { @@ -142,7 +143,7 @@ namespace Spring.Messaging.Ems.Support /// Creates the connection via the ConnectionFactory. /// /// - protected virtual Connection CreateConnection() + protected virtual IConnection CreateConnection() { return ConnectionFactory.CreateConnection(); } @@ -152,7 +153,7 @@ namespace Spring.Messaging.Ems.Support /// /// The connection to create a session for. /// The new session - protected virtual Session CreateSession(Connection con) + protected virtual ISession CreateSession(IConnection con) { return con.CreateSession(sessionTransacted, SessionAcknowledgeMode); } @@ -162,7 +163,7 @@ namespace Spring.Messaging.Ems.Support /// /// The session to check. /// true if in client ack mode, false otherwise - protected virtual bool IsClientAcknowledge(Session session) + protected virtual bool IsClientAcknowledge(ISession session) { return (session.AcknowledgeMode == Session.CLIENT_ACKNOWLEDGE); } diff --git a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsUtils.cs b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsUtils.cs index a123096c..6692c3df 100644 --- a/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsUtils.cs +++ b/src/Spring/Spring.Messaging.Ems/Messaging/Ems/Support/EmsUtils.cs @@ -20,6 +20,7 @@ using System; using Common.Logging; +using Spring.Messaging.Ems.Common; using Spring.Util; using TIBCO.EMS; @@ -42,7 +43,7 @@ namespace Spring.Messaging.Ems.Support /// /// the EMS Connection to close (may be null) /// - public static void CloseConnection(Connection con) + public static void CloseConnection(IConnection con) { CloseConnection(con, false); } @@ -54,7 +55,7 @@ namespace Spring.Messaging.Ems.Support /// /// whether to call stop() before closing /// - public static void CloseConnection(Connection con, bool stop) + public static void CloseConnection(IConnection con, bool stop) { if (con != null) { @@ -93,7 +94,7 @@ namespace Spring.Messaging.Ems.Support /// /// the EMS Session to close (may be null) /// - public static void CloseSession(Session session) + public static void CloseSession(ISession session) { if (session != null) { @@ -118,7 +119,7 @@ namespace Spring.Messaging.Ems.Support /// /// the EMS MessageProducer to close (may be null) /// - public static void CloseMessageProducer(MessageProducer producer) + public static void CloseMessageProducer(IMessageProducer producer) { if (producer != null) { @@ -143,7 +144,7 @@ namespace Spring.Messaging.Ems.Support /// /// the EMS MessageConsumer to close (may be null) /// - public static void CloseMessageConsumer(MessageConsumer consumer) + public static void CloseMessageConsumer(IMessageConsumer consumer) { if (consumer != null) { @@ -195,7 +196,7 @@ namespace Spring.Messaging.Ems.Support /// the EMS Session to commit /// /// EMSException if committing failed - public static void CommitIfNecessary(Session session) + public static void CommitIfNecessary(ISession session) { AssertUtils.ArgumentNotNull(session, "Session must not be null"); @@ -221,7 +222,7 @@ namespace Spring.Messaging.Ems.Support /// the EMS Session to rollback /// /// EMSException if committing failed - public static void RollbackIfNecessary(Session session) + public static void RollbackIfNecessary(ISession session) { AssertUtils.ArgumentNotNull(session, "Session must not be null"); session.Rollback(); @@ -260,5 +261,35 @@ namespace Spring.Messaging.Ems.Support } } } + + /// + /// Converts the acknowledgement mode from an integer to an enumeration. If the integer + /// does not match a valid enumeration, the returned enumeration is SessionMode.AutoAcknowledge + /// + /// The ack mode. + /// The corresponding SessionMode enumeration + public static SessionMode ConvertAcknowledgementMode(int ackMode) + { + switch (ackMode) + { + case Session.AUTO_ACKNOWLEDGE: + return SessionMode.AutoAcknowledge; + case Session.CLIENT_ACKNOWLEDGE: + return SessionMode.ClientAcknowledge; + case Session.DUPS_OK_ACKNOWLEDGE: + return SessionMode.DupsOkAcknowledge; + case Session.EXPLICIT_CLIENT_ACKNOWLEDGE: + return SessionMode.ExplicitClientAcknowledge; + case Session.EXPLICIT_CLIENT_DUPS_OK_ACKNOWLEDGE: + return SessionMode.ExplicitClientDupsOkAcknowledge; + case Session.NO_ACKNOWLEDGE: + return SessionMode.NoAcknowledge; + case Session.SESSION_TRANSACTED: + return SessionMode.SessionTransacted; + default: + logger.Warn("Integer acknowledgement mode [" + ackMode + "] not valid. Defaulting to SessionMode.AutoAcknowledge"); + return SessionMode.AutoAcknowledge; + } + } } } diff --git a/test/Spring/Spring.Messaging.Ems.Tests/Messaging/Ems/Integration/SimpleMessageListenerContainerTests.cs b/test/Spring/Spring.Messaging.Ems.Tests/Messaging/Ems/Integration/SimpleMessageListenerContainerTests.cs index cd58cb97..8ee0030b 100644 --- a/test/Spring/Spring.Messaging.Ems.Tests/Messaging/Ems/Integration/SimpleMessageListenerContainerTests.cs +++ b/test/Spring/Spring.Messaging.Ems.Tests/Messaging/Ems/Integration/SimpleMessageListenerContainerTests.cs @@ -23,9 +23,11 @@ using System; using System.Threading; using NUnit.Framework; +using Spring.Messaging.Ems.Common; using Spring.Messaging.Ems.Core; using Spring.Messaging.Ems.Listener; using Spring.Testing.NUnit; +using TIBCO.EMS.ADMIN; #endregion @@ -49,7 +51,9 @@ namespace Spring.Messaging.Ems.Integration SimpleMessageListener listener = applicationContext["SimpleMessageListener"] as SimpleMessageListener; Assert.IsNotNull(container); Assert.IsNotNull(listener); - + + Admin admin = new Admin("tcp://localhost:7222", "admin", null); + admin.PurgeQueue("test.queue"); EmsTemplate emsTemplate = (EmsTemplate) applicationContext["MessageTemplate"] as EmsTemplate; Assert.IsNotNull(emsTemplate); diff --git a/test/Spring/Spring.Messaging.Ems.Tests/Messaging/Ems/Integration/SimpleMessageListenerContainerTests.xml b/test/Spring/Spring.Messaging.Ems.Tests/Messaging/Ems/Integration/SimpleMessageListenerContainerTests.xml index cfd1033c..7ec18ce6 100644 --- a/test/Spring/Spring.Messaging.Ems.Tests/Messaging/Ems/Integration/SimpleMessageListenerContainerTests.xml +++ b/test/Spring/Spring.Messaging.Ems.Tests/Messaging/Ems/Integration/SimpleMessageListenerContainerTests.xml @@ -2,7 +2,10 @@ - + + + + diff --git a/test/Spring/Spring.Messaging.Ems.Tests/Spring.Messaging.Ems.Tests.2005.csproj b/test/Spring/Spring.Messaging.Ems.Tests/Spring.Messaging.Ems.Tests.2005.csproj index d9eac0b7..1cb30d2d 100644 --- a/test/Spring/Spring.Messaging.Ems.Tests/Spring.Messaging.Ems.Tests.2005.csproj +++ b/test/Spring/Spring.Messaging.Ems.Tests/Spring.Messaging.Ems.Tests.2005.csproj @@ -45,6 +45,10 @@ False ..\..\..\lib\Net\2.0\TIBCO.EMS.dll + + False + ..\..\..\lib\Net\2.0\TIBCO.EMS.ADMIN.dll +