Add interfaces for ConnectionFactory, Connection, Session, MessageConsumer, MessageProducer, and TopicSubscriber and EMS implementations

Add caching infrastructure
This commit is contained in:
markpollack
2009-05-03 04:55:39 +00:00
parent 27b6376f6b
commit c32eeb0727
51 changed files with 3612 additions and 200 deletions

View File

@@ -0,0 +1,131 @@
#region License
/*
* Copyright <20> 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);
}
}
}
}

View File

@@ -0,0 +1,217 @@
#region License
/*
* Copyright <20> 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
}
}

View File

@@ -0,0 +1,78 @@
#region License
/*
* Copyright <20> 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
}
}

View File

@@ -0,0 +1,119 @@
#region License
/*
* Copyright <20> 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
}
}

View File

@@ -0,0 +1,216 @@
#region License
/*
* Copyright <20> 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
}
}

View File

@@ -0,0 +1,46 @@
#region License
/*
* Copyright <20> 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
}
}

View File

@@ -0,0 +1,52 @@
#region License
/*
* Copyright <20> 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();
}
}

View File

@@ -0,0 +1,65 @@
#region License
/*
* Copyright <20> 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; }
}
}

View File

@@ -0,0 +1,37 @@
#region License
/*
* Copyright <20> 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();
}
}

View File

@@ -0,0 +1,52 @@
#region License
/*
* Copyright <20> 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; }
/// <summary>
/// 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.
/// </summary>
/// <remarks>Time to live is set to zero by default.</remarks>
/// <value>The message time to live in milliseconds; zero is unlimited</value>
long TimeToLive { get; set; }
}
}

View File

@@ -0,0 +1,77 @@
#region License
/*
* Copyright <20> 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; }
}
}

View File

@@ -0,0 +1,30 @@
#region License
/*
* Copyright <20> 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; }
}
}

View File

@@ -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)

View File

@@ -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
{
/// <summary>
/// NMS MessageConsumer decorator that adapts all calls
/// to a shared MessageConsumer instance underneath.
/// </summary>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET) </author>
public class CachedMessageConsumer : IMessageConsumer
{
private IMessageConsumer target;
/// <summary>
/// Initializes a new instance of the <see cref="CachedMessageConsumer"/> class.
/// </summary>
/// <param name="target">The target.</param>
public CachedMessageConsumer(IMessageConsumer target)
{
this.target = target;
}
/// <summary>
/// Gets the target MessageConsumer, the consumer we are 'wrapping'
/// </summary>
/// <value>The target MessageConsumer.</value>
public IMessageConsumer Target
{
get { return target; }
}
/// <summary>
/// Occurs when a message is received.
/// </summary>
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; }
}
/// <summary>
/// Receives the next message produced for this message consumer.
/// </summary>
/// <returns>the next message produced for this message consumer, , or null if this message consumer is concurrently closed</returns>
public Message Receive()
{
return this.target.Receive();
}
/// <summary>
/// Receives the next message that arrives within the specified timeout interval.
/// </summary>
/// <param name="timeout">The timeout value.</param>
/// <returns>the next message produced for this message consumer, or null if the timeout expires or this message consumer is concurrently closed</returns>
public Message Receive(long timeout)
{
return this.target.Receive(timeout);
}
/// <summary>
/// Receives the next message if one is immediately available.
/// </summary>
/// <returns>the next message produced for this message consumer, or null if one is not available</returns>
public Message ReceiveNoWait()
{
return this.target.ReceiveNoWait();
}
/// <summary>
/// No-op implementation since it is caching.
/// </summary>
public void Close()
{
// It's a cached MessageConsumer...
}
/// <summary>
/// Description that shows this is a cached MessageConsumer
/// </summary>
/// <returns>Description that shows this is a cached MessageConsumer</returns>
public override string ToString()
{
return "Cached EMS MessageConsumer: " + this.target;
}
}
}

View File

@@ -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
{
/// <summary>
/// MessageProducer decorator that adapts calls to a shared MessageProducer
/// instance underneath, managing QoS settings locally within the decorator.
/// </summary>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public class CachedMessageProducer : IMessageProducer
{
private readonly IMessageProducer target;
private object originalDisableMessageID;
private object originalDisableMessageTimestamp;
private MessageDeliveryMode deliveryMode;
private int priority;
private long timeToLive;
/// <summary>
/// Initializes a new instance of the <see cref="CachedMessageProducer"/> class.
/// </summary>
/// <param name="target">The target.</param>
public CachedMessageProducer(IMessageProducer target)
{
this.target = target;
deliveryMode = target.MsgDeliveryMode;
priority = target.Priority;
timeToLive = target.TimeToLive;
}
/// <summary>
/// Gets or sets a value indicating whether disable setting of the message ID property.
/// </summary>
/// <value><c>true</c> if disable message ID setting; otherwise, <c>false</c>.</value>
public bool DisableMessageID
{
get
{
return target.DisableMessageID;
}
set
{
if (originalDisableMessageID == null)
{
originalDisableMessageID = target.DisableMessageID;
}
target.DisableMessageID = value;
}
}
/// <summary>
/// Gets or sets a value indicating whether disable setting the message timestamp property.
/// </summary>
/// <value>
/// <c>true</c> if disable message timestamp; otherwise, <c>false</c>.
/// </value>
public bool DisableMessageTimestamp
{
get
{
return target.DisableMessageTimestamp;
}
set
{
if (originalDisableMessageTimestamp == null)
{
originalDisableMessageTimestamp = target.DisableMessageTimestamp;
}
target.DisableMessageTimestamp = value;
}
}
/// <summary>
/// Gets or sets the producer's default delivery mode.
/// </summary>
/// <value>The message delivery mode for this message producer</value>
public int DeliveryMode
{
get { return (int)this.deliveryMode; }
set { this.deliveryMode = (MessageDeliveryMode) value; }
}
/// <summary>
/// Gets or sets the MSG delivery mode.
/// </summary>
/// <value>The MSG delivery mode.</value>
public MessageDeliveryMode MsgDeliveryMode
{
get { return this.deliveryMode; }
set { this.deliveryMode = value; }
}
/// <summary>
/// Gets or sets the priority of messages sent with this producer.
/// </summary>
/// <value>The priority.</value>
public int Priority
{
get { return priority; }
set { priority = value;}
}
/// <summary>
/// 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.
/// </summary>
/// <remarks>Time to live is set to zero by default.</remarks>
/// <value>The message time to live in milliseconds; zero is unlimited</value>
public long TimeToLive
{
get { return timeToLive; }
set { timeToLive = value; }
}
public Destination Destination
{
get { return target.Destination; }
}
/// <summary>
/// Gets the target MessageProducer, the producer we are 'wrapping'
/// </summary>
/// <value>The target MessageProducer.</value>
public IMessageProducer Target
{
get { return target; }
}
/// <summary>
/// Sends the specified message.
/// </summary>
/// <param name="message">The message.</param>
public void Send(Message message)
{
target.Send(message, this.deliveryMode, this.priority, this.timeToLive);
}
/// <summary>
/// Sends the specified message.
/// </summary>
/// <param name="message">The message.</param>
/// <param name="deliveryMode">The delivery mode.</param>
/// <param name="priority">The priority.</param>
/// <param name="timeToLive">The time to live.</param>
public void Send(Message message, int deliveryMode, int priority, long timeToLive)
{
target.Send(message, deliveryMode, priority, timeToLive);
}
/// <summary>
/// Sends a message to the specified destination.
/// </summary>
/// <param name="destination">The destination.</param>
/// <param name="message">The message.</param>
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);
}
/// <summary>
/// Reset properties.
/// </summary>
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(); }
}
/// <summary>
/// Returns string indicated this is a wrapped MessageProducer
/// </summary>
/// <returns></returns>
public override string ToString()
{
return "Cached EMS MessageProducer: " + this.target;
}
}
}

View File

@@ -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
{
/// <summary>
/// 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).
/// </summary>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack</author>
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;
/// <summary>
/// Initializes a new instance of the <see cref="CachedSession"/> class.
/// </summary>
/// <param name="targetSession">The target session.</param>
/// <param name="sessionList">The session list.</param>
/// <param name="ccf">The CachingConnectionFactory.</param>
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;
}
/// <summary>
/// Gets the target, for testing purposes.
/// </summary>
/// <value>The target.</value>
public ISession TargetSession
{
get { return target; }
}
/// <summary>
/// Creates the producer, potentially returning a cached instance.
/// </summary>
/// <param name="destination">The destination.</param>
/// <returns>A message producer.</returns>
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);
}
}
/// <summary>
/// If have not yet reached session cache size, cache the session, otherwise
/// dispose of all cached message producers and close the session.
/// </summary>
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();
}
}
/// <summary>
/// Creates the consumer, potentially returning a cached instance.
/// </summary>
/// <param name="destination">The destination.</param>
/// <returns>A message consumer</returns>
public IMessageConsumer CreateConsumer(Destination destination)
{
return CreateConsumer(destination, null, false, null);
}
/// <summary>
/// Creates the consumer, potentially returning a cached instance.
/// </summary>
/// <param name="destination">The destination.</param>
/// <param name="selector">The selector.</param>
/// <returns>A message consumer</returns>
public IMessageConsumer CreateConsumer(Destination destination, string selector)
{
return CreateConsumer(destination, selector, false, null);
}
/// <summary>
/// Creates the consumer, potentially returning a cached instance.
/// </summary>
/// <param name="destination">The destination.</param>
/// <param name="selector">The selector.</param>
/// <param name="noLocal">if set to <c>true</c> [no local].</param>
/// <returns>A message consumer.</returns>
public IMessageConsumer CreateConsumer(Destination destination, string selector, bool noLocal)
{
return CreateConsumer(destination, selector, noLocal, null);
}
/// <summary>
/// Creates the durable consumer, potentially returning a cached instance.
/// </summary>
/// <param name="destination">The destination.</param>
/// <param name="subscription">The name of the durable subscription.</param>
/// <param name="selector">The selector.</param>
/// <param name="noLocal">if set to <c>true</c> [no local].</param>
/// <returns>A message consumer</returns>
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);
}
}
/// <summary>
/// Creates the durable consumer, potentially returning a cached instance.
/// </summary>
/// <param name="destination">The destination.</param>
/// <param name="subscription">The name of the durable subscription.</param>
/// <returns>A message consumer</returns>
public ITopicSubscriber CreateDurableSubscriber(Topic destination, string subscription)
{
return CreateDurableSubscriber(destination, subscription, null, false);
}
/// <summary>
/// Deletes the durable consumer.
/// </summary>
/// <param name="name">The name.</param>
public void DeleteDurableConsumer(string name)
{
throw new NotImplementedException();
}
/// <summary>
/// Deletes the durable consumer.
/// </summary>
/// <param name="name">The name.</param>
/// <param name="requestTimeout">The request timeout.</param>
public void DeleteDurableConsumer(string name, TimeSpan requestTimeout)
{
throw new NotImplementedException();
}
/// <summary>
/// Creates the consumer.
/// </summary>
/// <param name="destination">The destination.</param>
/// <param name="selector">The selector.</param>
/// <param name="noLocal">if set to <c>true</c> [no local].</param>
/// <param name="subscription">The subscription.</param>
/// <returns></returns>
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
/// <summary>
/// Gets the queue.
/// </summary>
/// <param name="name">The name.</param>
/// <returns></returns>
public TIBCO.EMS.Queue CreateQueue(string name)
{
this.transactionOpen = true;
return target.CreateQueue(name);
}
/// <summary>
/// Gets the topic.
/// </summary>
/// <param name="name">The name.</param>
/// <returns></returns>
public Topic CreateTopic(string name)
{
this.transactionOpen = true;
return target.CreateTopic(name);
}
/// <summary>
/// Creates the temporary queue.
/// </summary>
/// <returns></returns>
public TemporaryQueue CreateTemporaryQueue()
{
this.transactionOpen = true;
return target.CreateTemporaryQueue();
}
/// <summary>
/// Creates the temporary topic.
/// </summary>
/// <returns></returns>
public TemporaryTopic CreateTemporaryTopic()
{
this.transactionOpen = true;
return target.CreateTemporaryTopic();
}
/// <summary>
/// Creates the message.
/// </summary>
/// <returns></returns>
public Message CreateMessage()
{
this.transactionOpen = true;
return target.CreateMessage();
}
/// <summary>
/// Creates the text message.
/// </summary>
/// <returns></returns>
public TextMessage CreateTextMessage()
{
this.transactionOpen = true;
return target.CreateTextMessage();
}
/// <summary>
/// Creates the text message.
/// </summary>
/// <param name="text">The text.</param>
/// <returns></returns>
public TextMessage CreateTextMessage(string text)
{
this.transactionOpen = true;
return target.CreateTextMessage(text);
}
/// <summary>
/// Creates the map message.
/// </summary>
/// <returns></returns>
public MapMessage CreateMapMessage()
{
this.transactionOpen = true;
return target.CreateMapMessage();
}
/// <summary>
/// Creates the bytes message.
/// </summary>
/// <returns></returns>
public BytesMessage CreateBytesMessage()
{
this.transactionOpen = true;
return target.CreateBytesMessage();
}
/// <summary>
/// Creates the object message.
/// </summary>
/// <returns></returns>
public ObjectMessage CreateObjectMessage()
{
this.transactionOpen = true;
return target.CreateObjectMessage();
}
/// <summary>
/// Creates the object message.
/// </summary>
/// <param name="body">The body.</param>
/// <returns></returns>
public ObjectMessage CreateObjectMessage(object body)
{
this.transactionOpen = true;
return target.CreateObjectMessage(body);
}
/// <summary>
/// Creates the stream message.
/// </summary>
/// <returns></returns>
public StreamMessage CreateStreamMessage()
{
this.transactionOpen = true;
return target.CreateStreamMessage();
}
/// <summary>
/// Commits this instance.
/// </summary>
public void Commit()
{
this.transactionOpen = false;
target.Commit();
}
/// <summary>
/// Rollbacks this instance.
/// </summary>
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);
}
/// <summary>
/// Gets a value indicating whether this <see cref="CachedSession"/> is transacted.
/// </summary>
/// <value><c>true</c> if transacted; otherwise, <c>false</c>.</value>
public bool Transacted
{
get
{
this.transactionOpen = true;
return target.Transacted;
}
}
/// <summary>
/// Gets the acknowledgement mode.
/// </summary>
/// <value>The acknowledgement mode.</value>
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
/// <summary>
/// Returns a <see cref="T:System.String"/> that represents the current <see cref="T:System.Object"/>.
/// </summary>
/// <returns>
/// A <see cref="T:System.String"/> that represents the current <see cref="T:System.Object"/>.
/// </returns>
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();
}
}
}

View File

@@ -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
{
/// <summary>
/// <see cref="SingleConnectionFactory"/> 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.
/// </summary>
/// <remarks>
/// 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.
/// <para>
/// 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.
/// </para>
/// <para>
/// 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 <code>Session.Close()</code> 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.
/// </para>
/// </remarks>
///
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
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();
/// <summary>
/// Initializes a new instance of the <see cref="CachingConnectionFactory"/> class.
/// and sets the ReconnectOnException to true
/// </summary>
public CachingConnectionFactory()
{
ReconnectOnException = true;
}
/// <summary>
/// Initializes a new instance of the <see cref="CachingConnectionFactory"/> class for the given
/// IConnectionFactory
/// </summary>
/// <param name="targetConnectionFactory">The target connection factory.</param>
public CachingConnectionFactory(IConnectionFactory targetConnectionFactory) : base(targetConnectionFactory)
{
ReconnectOnException = true;
}
/// <summary>
/// Gets or sets the size of the session cache.
/// </summary>
/// <remarks>
/// 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.
/// <para>
/// 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.
/// </para>
/// </remarks>
/// <value>The size of the session cache.</value>
public int SessionCacheSize
{
get { return sessionCacheSize; }
set
{
AssertUtils.IsTrue(value >= 1, "Session cache size must be 1 or higher");
sessionCacheSize = value;
}
}
/// <summary>
/// Gets or sets a value indicating whether to cache MessageProducers per
/// Session instance. (more specifically: one MessageProducer per Destination
/// and Session).
/// </summary>
/// <remarks>
/// <para>Default is "true". Switch this to "false" in order to always,
/// recreate MessageProducers on demand.
/// </para>
/// </remarks>
/// <value><c>true</c> if should cache message producers; otherwise, <c>false</c>.</value>
public bool CacheProducers
{
get { return cacheProducers; }
set { cacheProducers = value; }
}
/// <summary>
/// Gets or sets a value indicating whether o cache JMS MessageConsumers per
/// NMS Session instance.
/// </summary>
/// <remarks>
/// 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.
/// <para>
/// Default is "true". Switch this to "false" in order to always
/// recreate MessageConsumers on demand.
/// </para>
/// </remarks>
/// <value><c>true</c> to cache consumers per session instance; otherwise, <c>false</c>.</value>
public bool CacheConsumers
{
get { return cacheConsumers; }
set { cacheConsumers = value; }
}
/// <summary>
/// Gets or sets a value indicating whether this instance is active.
/// </summary>
/// <value><c>true</c> if this instance is active; otherwise, <c>false</c>.</value>
public bool IsActive
{
get { return active; }
set { active = value; }
}
/// <summary>
/// Resets the Session cache as well as resetting the connection.
/// </summary>
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();
}
/// <summary>
/// Obtaining a cached Session.
/// </summary>
/// <param name="con">The connection to operate on.</param>
/// <param name="mode">The session ack mode.</param>
/// <returns>The Session to use
/// </returns>
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);
}
/// <summary>
/// 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.
/// </summary>
/// <param name="targetSession">The original Session to wrap.</param>
/// <param name="sessionList">The List of cached Sessions that the given Session belongs to.</param>
/// <returns>The wrapped Session</returns>
protected virtual ISession GetCachedSessionWrapper(ISession targetSession, LinkedList sessionList)
{
return new CachedSession(targetSession, sessionList, this);
}
}
}

View File

@@ -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
/// <summary>
/// Releases the given connection.
/// Releases the given connection, stopping it (if necessary) and eventually closing it.
/// </summary>
/// <remarks>Checks <see cref="ISmartConnectionFactory.ShouldStop"/>, if available.
/// This is essentially a more sophisticated version of
/// <see cref="EmsUtils.CloseConnection(IConnection, bool)"/>
/// </remarks>
/// <param name="connection">The connection to release. (if this is <code>null</code>, the call will be ignored)</param>
/// <param name="cf">The ConnectionFactory that the Connection was obtained from. (may be <code>null</code>)</param>
/// <param name="started">whether the Connection might have been started by the application.</param>
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);
}
}
}
/// <summary>
/// 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.
/// </summary>
/// <param name="session">The session to unwrap</param>
/// <returns>The innermost target Session, or the passed-in one if no decorator</returns>
public static ISession GetTargetSession(ISession session)
{
ISession sessionToUse = session;
while (sessionToUse is IDecoratorSession)
{
sessionToUse = ((IDecoratorSession)sessionToUse).TargetSession;
}
return sessionToUse;
}
/// <summary>
@@ -71,7 +106,7 @@ namespace Spring.Messaging.Ems.Connections
/// <returns>
/// <c>true</c> if is session transactional, bound to current thread; otherwise, <c>false</c>.
/// </returns>
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
/// <returns> the transactional Session, or <code>null</code> if none found
/// </returns>
/// <throws> EMSException in case of EMS failure </throws>
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 <code>null</code> if none found
/// </returns>
/// <throws>EMSException in case of EMS failure </throws>
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
/// <returns> an appropriate Session fetched from the holder,
/// or <code>null</code> if none found
/// </returns>
Session GetSession(EmsResourceHolder holder);
ISession GetSession(EmsResourceHolder holder);
/// <summary> Fetch an appropriate Connection from the given EmsResourceHolder.</summary>
/// <param name="holder">the EmsResourceHolder
@@ -276,13 +311,13 @@ namespace Spring.Messaging.Ems.Connections
/// <returns> an appropriate Connection fetched from the holder,
/// or <code>null</code> if none found
/// </returns>
Connection GetConnection(EmsResourceHolder holder);
IConnection GetConnection(EmsResourceHolder holder);
/// <summary> Create a new EMS Connection for registration with a EmsResourceHolder.</summary>
/// <returns> the new EMS Connection
/// </returns>
/// <throws>EMSException if thrown by EMS API methods </throws>
Connection CreateConnection();
IConnection CreateConnection();
/// <summary> Create a new EMS Session for registration with a EmsResourceHolder.</summary>
/// <param name="con">the EMS Connection to create a Session for
@@ -290,7 +325,7 @@ namespace Spring.Messaging.Ems.Connections
/// <returns> the new EMS Session
/// </returns>
/// <throws>EMSException if thrown by EMS API methods </throws>
Session CreateSession(Connection con);
ISession CreateSession(IConnection con);
/// <summary>

View File

@@ -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
/// <param name="connectionFactory">The connection factory that this
/// resource holder is associated with (may be <code>null</code>)
/// </param>
public EmsResourceHolder(ConnectionFactory connectionFactory)
public EmsResourceHolder(IConnectionFactory connectionFactory)
{
this.connectionFactory = connectionFactory;
}
@@ -85,7 +86,7 @@ namespace Spring.Messaging.Ems.Connections
/// given Session.
/// </summary>
/// <param name="session">The session.</param>
public EmsResourceHolder(Session session)
public EmsResourceHolder(ISession session)
{
AddSession(session);
frozen = true;
@@ -96,7 +97,7 @@ namespace Spring.Messaging.Ems.Connections
/// </param>
/// <param name="session">the EMS Session
/// </param>
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
/// <param name="connectionFactory">The connection factory.</param>
/// <param name="connection">The connection.</param>
/// <param name="session">The session.</param>
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.
/// </summary>
/// <param name="connection">The connection.</param>
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.
/// </summary>
/// <param name="session">The session.</param>
public void AddSession(Session session)
public void AddSession(ISession session)
{
AddSession(session, null);
}
@@ -166,7 +167,7 @@ namespace Spring.Messaging.Ems.Connections
/// </summary>
/// <param name="session">The session.</param>
/// <param name="connection">The connection.</param>
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
/// </summary>
/// <returns>A Connection, or null if no managed connection.</returns>
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);
}
/// <summary>
@@ -202,18 +203,18 @@ namespace Spring.Messaging.Ems.Connections
/// </summary>
/// <param name="connectionType">Type of the connection.</param>
/// <returns>The connection, or null if not found.</returns>
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);
}
/// <summary>
/// Gets the first session manged by this holder or null if not available.
/// </summary>
/// <returns>The session or null if not available.</returns>
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);
}
/// <summary>
@@ -221,7 +222,7 @@ namespace Spring.Messaging.Ems.Connections
/// </summary>
/// <param name="sessionType">Type of the session.</param>
/// <returns>The session or null if not available.</returns>
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
/// <param name="sessionType">Type of the session.</param>
/// <param name="connection">The connection.</param>
/// <returns>The sessin or null if not available.</returns>
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);
}
/// <summary>
@@ -247,7 +248,7 @@ namespace Spring.Messaging.Ems.Connections
/// </summary>
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
/// </summary>
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
/// <returns>
/// <c>true</c> if the holder contains the specified session; otherwise, <c>false</c>.
/// </returns>
public bool ContainsSession(Session session)
public bool ContainsSession(ISession session)
{
return this.sessions.Contains(session);
}

View File

@@ -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;
/// <summary>
/// Initializes a new instance of the <see cref="EmsTransactionManager"/> class.
@@ -86,7 +87,7 @@ namespace Spring.Messaging.Ems.Connections
/// given a ConnectionFactory.
/// </summary>
/// <param name="connectionFactory">The connection factory to obtain connections from.</param>
public EmsTransactionManager(ConnectionFactory connectionFactory) : this()
public EmsTransactionManager(IConnectionFactory connectionFactory) : this()
{
ConnectionFactory = connectionFactory;
AfterPropertiesSet();
@@ -98,7 +99,7 @@ namespace Spring.Messaging.Ems.Connections
/// for.
/// </summary>
/// <value>The connection factory.</value>
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
/// </summary>
/// <returns>The new Connection</returns>
/// <exception cref="EMSException">If thrown by underlying messaging APIs</exception>
protected virtual Connection CreateConnection()
protected virtual IConnection CreateConnection()
{
return ConnectionFactory.CreateConnection();
}
@@ -384,7 +385,7 @@ namespace Spring.Messaging.Ems.Connections
/// <param name="connection">The connection to create a Session for.</param>
/// <returns>the new Session</returns>
/// <exception cref="EMSException">If thrown by underlying messaging APIs</exception>
protected virtual Session CreateSession(Connection connection)
protected virtual ISession CreateSession(IConnection connection)
{
return connection.CreateSession(true, Session.SESSION_TRANSACTED);
}

View File

@@ -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
{
/// <summary>
/// Subinterface of Session to be implemented by
/// implementations that wrap an Session to provide added
/// functionality. Allows access to the the underlying target Session.
/// </summary>
/// <author>Mark Pollack</author>
/// <see cref="ConnectionFactoryUtils.GetTargetSession(ISession)"/>
/// <see cref="CachingConnectionFactory"/>
public interface IDecoratorSession : ISession
{
/// <summary>
/// Gets the target session of the decorator.
/// This will typically be the native provider Session or a wrapper from a session pool.
/// </summary>
/// <value>The underlying session, never null</value>
ISession TargetSession { get; }
}
}

View File

@@ -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
{
/// <summary>
/// Extension of the <code>IConnectionFactory</code> interface,
/// indicating how to release Connections obtained from it.
/// </summary>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack (.NET)</author>
public interface ISmartConnectionFactory
{
/// <summary>
/// Shoulds we stop the connection, obtained from this ConnectionFactory?
/// </summary>
/// <param name="con">The connection to check.</param>
/// <returns>wheter a stop call is necessary</returns>
bool ShouldStop(IConnection con);
}
}

View File

@@ -0,0 +1,777 @@
#region License
/*
* Copyright <20> 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
{
/// <summary>
/// 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.
/// </summary>
/// <remarks>
/// <para>
/// You can either pass in a specific Connection directly or let this
/// factory lazily create a Connection via a given target ConnectionFactory.
/// </para>
/// <para>
/// Useful for testing and in applications when you want to keep using the
/// same Connection for multiple <see cref="EmsTemplate"/>
/// calls, without having a pooling ConnectionFactory underneath. This may span
/// any number of transactions, even concurrently executing transactions.
/// </para>
/// <para>
/// 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.
/// </para>
/// </remarks>
/// <author>Juergen Hoeller</author>
/// <author>Mark Pollack</author>
/// <author>Mark Pollack (.NET)</author>
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;
/// <summary>
/// Wrapped Connection
/// </summary>
private IConnection target;
/// <summary>
/// Proxy Connection
/// </summary>
private IConnection connection;
/// <summary>
/// Whether the shared Connection has been started
/// </summary>
private bool started = false;
/// <summary>
/// Synchronization monitor for the shared Connection
/// </summary>
private object connectionMonitor = new object();
#endregion
#region Constructors
/// <summary>
/// Initializes a new instance of the <see cref="SingleConnectionFactory"/> class.
/// </summary>
public SingleConnectionFactory()
{
}
/// <summary>
/// Initializes a new instance of the <see cref="SingleConnectionFactory"/> class
/// that alwasy returns the given Connection.
/// </summary>
/// <param name="target">The single Connection.</param>
public SingleConnectionFactory(IConnection target)
{
AssertUtils.ArgumentNotNull(target, "connection", "TargetSession Connection must not be null");
this.target = target;
connection = GetSharedConnection(target);
}
/// <summary>
/// Initializes a new instance of the <see cref="SingleConnectionFactory"/> class
/// that alwasy returns a single Connection.
/// </summary>
/// <param name="targetConnectionFactory">The target connection factory.</param>
public SingleConnectionFactory(IConnectionFactory targetConnectionFactory)
{
AssertUtils.ArgumentNotNull(targetConnectionFactory, "targetConnectionFactory",
"TargetSession ConnectionFactory must not be null");
this.targetConnectionFactory = targetConnectionFactory;
}
#endregion
#region Properties
/// <summary>
/// Gets or sets the target connection factory which will be used to create a single
/// connection.
/// </summary>
/// <value>The target connection factory.</value>
public IConnectionFactory TargetConnectionFactory
{
get { return targetConnectionFactory; }
set { targetConnectionFactory = value; }
}
/// <summary>
/// Sets the exception listener.
/// </summary>
/// <value>The exception listener.</value>
public IExceptionListener ExceptionListener
{
set { exceptionListener = value; }
}
/// <summary>
/// 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.
/// </summary>
/// <remarks>
/// Default is <code>false</code>. Switch this to <code>true</code>
/// to automatically trigger recover based on your messaging provider's
/// exception notifications.
/// <para>
/// 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.
/// </para>
/// </remarks>
/// <value>
/// <c>true</c> attempt to reconnect on exception during next access; otherwise, <c>false</c>.
/// </value>
public bool ReconnectOnException
{
get { return reconnectOnException; }
set { reconnectOnException = value; }
}
/// <summary>
/// Gets the connection monitor.
/// </summary>
/// <value>The connection monitor.</value>
internal object ConnectionMonitor
{
get { return connectionMonitor; }
}
/// <summary>
/// Gets a value indicating whether this instance is started.
/// </summary>
/// <value>
/// <c>true</c> if this instance is started; otherwise, <c>false</c>.
/// </value>
internal bool IsStarted
{
get { return started;}
set { started = value; }
}
/// <summary>
/// Gets the client id.
/// </summary>
/// <value>The client id.</value>
internal string ClientId
{
get { return clientId; }
}
#endregion
#region Implementation of IConnectionFactory
/// <summary>
/// Creates the connection.
/// </summary>
/// <returns>A single shared connection</returns>
public IConnection CreateConnection()
{
lock (connectionMonitor)
{
if (connection == null)
{
InitConnection();
}
return connection;
}
}
/// <summary>
/// Creates the connection.
/// </summary>
/// <param name="userName">Name of the user.</param>
/// <param name="password">The password.</param>
/// <returns></returns>
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
/// <summary>
/// Initialize the underlying shared Connection. Closes and reinitializes the Connection if an underlying
/// Connection is present already.
/// </summary>
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);
}
}
/// <summary>
/// Exception listener callback that renews the underlying single Connection.
/// </summary>
/// <param name="exception">The exception from the messaging infrastructure.</param>
public void OnException(EMSException exception)
{
ResetConnection();
}
/// <summary>
/// Prepares the connection before it is exposed.
/// The default implementation applies ExceptionListener and client id.
/// Can be overridden in subclasses.
/// </summary>
/// <param name="con">The Connection to prepare.</param>
/// <exception cref="EMSException">if thrown by any NMS API methods.</exception>
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);
}*/
}
/// <summary>
/// Template method for obtaining a (potentially cached) Session.
/// </summary>
/// <param name="con">The connection to operate on.</param>
/// <param name="mode">The session ack mode.</param>
/// <returns>the Session to use, or <code>null</code> to indicate
/// creation of a raw standard Session</returns>
public virtual ISession GetSession(IConnection con, SessionMode mode)
{
return null;
}
/// <summary>
/// reate a JMS Connection via this template's ConnectionFactory.
/// </summary>
/// <returns></returns>
protected virtual IConnection DoCreateConnection()
{
return TargetConnectionFactory.CreateConnection();
}
/// <summary>
/// Closes the given connection.
/// </summary>
/// <param name="con">The connection.</param>
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
/// <summary>
/// Ensure that the connection or TargetConnectionFactory are specified.
/// </summary>
public void AfterPropertiesSet()
{
if (connection == null && TargetConnectionFactory == null)
{
throw new ArgumentException("Connection or 'TargetConnectionFactory' is required.");
}
}
#endregion
/// <summary>
/// Close the underlying shared connection. The provider of this ConnectionFactory needs to care for proper shutdown.
/// As this object implements <see cref="IDisposable"/> an application context will automatically
/// invoke this on distruction o
/// </summary>
public void Dispose()
{
ResetConnection();
}
/// <summary>
/// Resets the underlying shared Connection, to be reinitialized on next access.
/// </summary>
public virtual void ResetConnection()
{
lock (connectionMonitor)
{
if (this.target != null)
{
CloseConnection(this.target);
}
this.target = null;
this.connection = null;
}
}
/// <summary>
/// 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.
/// </summary>
/// <param name="target">The original connection to wrap.</param>
/// <returns>the wrapped connection</returns>
protected virtual IConnection GetSharedConnection(IConnection target)
{
lock (connectionMonitor)
{
return new CloseSupressingConnection(this, target);
}
}
}
/// <summary>
/// Internal chained ExceptionListener for handling the internal recovery listener
/// in combination with a user-specified listener.
/// </summary>
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;
}
/// <summary>
/// Add information to show this is a shared NMS connection
/// </summary>
/// <returns>Description of connection wrapper</returns>
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
}
}

View File

@@ -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
/// <summary>
/// Delegate callback for browsing the messages in an EMS queue.
/// </summary>
public delegate object BrowserDelegate(Session session, QueueBrowser browser);
public delegate object BrowserDelegate(ISession session, QueueBrowser browser);
}

View File

@@ -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.
/// </summary>
/// <value>The connection factory.</value>
public ConnectionFactory ConnectionFactory
public IConnectionFactory ConnectionFactory
{
get
{
@@ -83,7 +84,7 @@ namespace Spring.Messaging.Ems.Core
///
/// <param name="connectionFactory">The connection factory.</param>
/// <returns></returns>
protected virtual EmsTemplate CreateEmsTemplate(ConnectionFactory connectionFactory)
protected virtual EmsTemplate CreateEmsTemplate(IConnectionFactory connectionFactory)
{
return new EmsTemplate(connectionFactory);
}

View File

@@ -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
/// <summary> Create a new EmsTemplate, given a ConnectionFactory.</summary>
/// <param name="connectionFactory">the ConnectionFactory to obtain Connections from
/// </param>
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; }
}
/// <summary>
/// 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
/// </summary>
/// <remarks>Connection/Session/MessageProducer are thread-safe classes in TIBCO EMS.</remarks>
/// <value><c>true</c> to locally cache ems resources; otherwise, <c>false</c>.</value>
virtual public bool CacheEmsResources
/* virtual public bool CacheEmsResources
{
get { return cacheEmsResources; }
set { cacheEmsResources = value; }
}
}*/
#endregion
@@ -428,7 +436,7 @@ namespace Spring.Messaging.Ems.Core
/// <returns> an appropriate Connection fetched from the holder,
/// or <code>null</code> if none found
/// </returns>
protected virtual Connection GetConnection(EmsResourceHolder holder)
protected virtual IConnection GetConnection(EmsResourceHolder holder)
{
return holder.GetConnection();
}
@@ -440,7 +448,7 @@ namespace Spring.Messaging.Ems.Core
/// <returns> an appropriate Session fetched from the holder,
/// or <code>null</code> if none found
/// </returns>
protected virtual Session GetSession(EmsResourceHolder holder)
protected virtual ISession GetSession(EmsResourceHolder holder)
{
return holder.GetSession();
}
@@ -463,9 +471,9 @@ namespace Spring.Messaging.Ems.Core
/// </seealso>
/// <seealso cref="MessageTimestampEnabled">
/// </seealso>
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
/// <returns>
/// <c>true</c> if the session is locally transacted; otherwise, <c>false</c>.
/// </returns>
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
/// <returns> the new EMS MessageProducer
/// </returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
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);
}
}*/
}
/// <summary> Create a EMS MessageConsumer for the given Session and Destination.
@@ -567,8 +577,8 @@ namespace Spring.Messaging.Ems.Core
/// <returns> the new EMS MessageConsumer
/// </returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
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
/// <returns>A EMS Connection
/// </returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
protected override Connection CreateConnection()
/* protected override IConnection CreateConnection()
{
if (CacheEmsResources)
{
@@ -607,7 +617,7 @@ namespace Spring.Messaging.Ems.Core
{
return ConnectionFactory.CreateConnection();
}
}
}*/
/// <summary> Create a EMS Session for the given Connection.
/// </summary>
@@ -620,7 +630,7 @@ namespace Spring.Messaging.Ems.Core
/// <returns> the new EMS Session
/// </returns>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
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);
}
}
}*/
/// <summary>
/// Send the given message.
@@ -642,7 +652,7 @@ namespace Spring.Messaging.Ems.Core
/// <param name="session">The session to operate on.</param>
/// <param name="destination">The destination to send to.</param>
/// <param name="messageCreatorDelegate">The message creator delegate callback to create a Message.</param>
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
/// <param name="session">The session to operate on.</param>
/// <param name="destination">The destination to send to.</param>
/// <param name="messageCreator">The message creator callback to create a Message.</param>
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
/// <param name="messageCreatorDelegate">delegate callback to create a EMS Message
/// </param>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
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
/// <param name="message">the EMS Message to send
/// </param>
/// <exception cref="EMSException">If there is any problem accessing the EMS API</exception>
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
/// <param name="destination">The destination to receive from.</param>
/// <param name="messageSelector">The message selector for this consumer (can be <code>null</code></param>
/// <returns>The Message received, or <code>null</code> if none.</returns>
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
/// <param name="session">The session to operate on.</param>
/// <param name="consumer">The consumer to receive with.</param>
/// <returns>The Message received, or <code>null</code> if none</returns>
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
/// <param name="queue">The queue.</param>
/// <param name="selector">The selector.</param>
/// <returns>A new queue browser</returns>
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.
/// </summary>
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)
{

View File

@@ -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
/// <param name="browser">The browser.</param>
/// <returns>The object from working with the Session and QueueBrowser, may be null</returns>
/// <exception cref="EMSException">If there is any problem when accessing EMS API</exception>
object DoInEms(Session session, QueueBrowser browser);
object DoInEms(ISession session, QueueBrowser browser);
}
}

View File

@@ -18,7 +18,6 @@
#endregion
using Spring.Messaging.Ems.Core;
using TIBCO.EMS;
namespace Spring.Messaging.Ems.Core

View File

@@ -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
/// <returns> the <code>Message</code> to be sent
/// </returns>
/// <throws>EMSException if thrown by EMS API methods </throws>
Message CreateMessage(Session session);
Message CreateMessage(ISession session);
}
}

View File

@@ -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
/// </param>
/// <returns> a result object from working with the <code>Session</code>, if any (can be <code>null</code>)
/// </returns>
object DoInEms(Session session, MessageProducer producer);
object DoInEms(ISession session, IMessageProducer producer);
}
}

View File

@@ -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
/// <returns> a result object from working with the <code>Session</code>, if any (so can be <code>null</code>)
/// </returns>
/// <throws>EMSException if there is any problem </throws>
object DoInEms(Session session);
object DoInEms(ISession session);
}
}

View File

@@ -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
/// <returns> the <code>Message</code> to be sent
/// </returns>
/// <throws>EMSException if thrown by EMS API methods </throws>
public delegate Message MessageCreatorDelegate(Session session);
public delegate Message MessageCreatorDelegate(ISession session);
}

View File

@@ -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
/// </param>
/// <returns> a result object from working with the <code>Session</code>, if any (can be <code>null</code>)
/// </returns>
public delegate object ProducerDelegate(Session session, MessageProducer producer);
public delegate object ProducerDelegate(ISession session, IMessageProducer producer);
}

View File

@@ -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
/// </returns>
/// <throws>EMSException if there is any problem </throws>
/// <author>Mark Pollack</author>
public delegate object SessionDelegate(Session session);
public delegate object SessionDelegate(ISession session);
}

View File

@@ -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.
/// </exception>
/// <see cref="SharedConnectionEnabled"/>
protected Connection SharedConnection
protected IConnection SharedConnection
{
get
{
@@ -446,9 +447,9 @@ namespace Spring.Messaging.Ems.Listener
/// </remarks>
/// <returns>the prepared Connection</returns>
/// <exception cref="EMSException">if the creation failed.</exception>
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
/// </remarks>
/// <param name="connection">The connection to prepare.</param>
/// <exception cref="EMSException">If the preparation efforts failed.</exception>
protected virtual void PrepareSharedConnection(Connection connection)
protected virtual void PrepareSharedConnection(IConnection connection)
{
if (ClientId != null)
{

View File

@@ -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
/// <see cref="CommitIfNecessary"/>
/// <see cref="RollbackOnExceptionIfNecessary"/>
/// <see cref="HandleListenerException"/>
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
/// <see cref="InvokeListener"/>
/// <see cref="CommitIfNecessary"/>
/// <see cref="RollbackOnExceptionIfNecessary"/>
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
/// <param name="message">The received message.</param>
/// <exception cref="EMSException">If thrown by EMS API methods.</exception>
/// <see cref="MessageListener"/>
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
/// <exception cref="EMSException">If thrown by EMS API methods.</exception>
/// <see cref="ISessionAwareMessageListener"/>
/// <see cref="ExposeListenerSession"/>
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
/// <param name="session">The session to commit.</param>
/// <param name="message">The message to acknowledge.</param>
/// <exception cref="EMSException">In case of commit failure</exception>
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
/// <c>true</c> if the is session locally transacted; otherwise, <c>false</c>.
/// </returns>
/// <see cref="EmsAccessor.SessionTransacted"/>
protected virtual bool IsSessionLocallyTransacted(Session session)
protected virtual bool IsSessionLocallyTransacted(ISession session)
{
return SessionTransacted;
}
@@ -485,7 +486,7 @@ namespace Spring.Messaging.Ems.Listener
/// </summary>
/// <param name="session">The session to rollback.</param>
/// <exception cref="EMSException">In case of a rollback error</exception>
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
/// <param name="session">The session to rollback.</param>
/// <param name="ex">The thrown application exception.</param>
/// <exception cref="EMSException">in case of a rollback error.</exception>
protected virtual void RollbackOnExceptionIfNecessary(Session session, Exception ex)
protected virtual void RollbackOnExceptionIfNecessary(ISession session, Exception ex)
{
try
{

View File

@@ -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
/// </summary>
/// <param name="message">The incoming message.</param>
/// <param name="session">The session to operate on.</param>
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
/// <param name="result">The result object to handle (never <code>null</code>).</param>
/// <param name="request">The original request message.</param>
/// <param name="session">The session to operate on (may be <code>null</code>).</param>
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
/// <returns>the JMS <code>Message</code> (never <code>null</code>)</returns>
/// <exception cref="MessageConversionException">If there was an error in message conversion</exception>
/// <exception cref="EMSException">if thrown by EMS API methods</exception>
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
/// <returns>the response destination (never <code>null</code>)</returns>
/// <exception cref="EMSException">if thrown by EMS API methods</exception>
/// <exception cref="InvalidDestinationException">if no destination can be determined.</exception>
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
/// </summary>
/// <param name="session">The session to operate on.</param>
/// <returns>The located destination</returns>
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
/// <param name="session">The session to operate on.</param>
/// <param name="destination">The destination to send to.</param>
/// <param name="response">The outgoing message about to be sent.</param>
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
/// </summary>
/// <param name="producer">The producer that will be used to send the message.</param>
/// <param name="response">The outgoing message about to be sent.</param>
protected virtual void PostProcessProducer(MessageProducer producer, Message response)
protected virtual void PostProcessProducer(IMessageProducer producer, Message response)
{
}

View File

@@ -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
/// <param name="session">the underlying EMS Session
/// </param>
/// <throws> EMSException if thrown by EMS methods </throws>
void OnMessage(Message message, Session session);
void OnMessage(Message message, ISession session);
}
}

View File

@@ -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 <see cref="LocallyExposedEmsResourceHolder"/> class.
/// </summary>
/// <param name="session">The session.</param>
public LocallyExposedEmsResourceHolder(Session session) : base(session)
public LocallyExposedEmsResourceHolder(ISession session) : base(session)
{
}

View File

@@ -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.
/// </summary>
/// <param name="connection"></param>
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
/// <param name="session">The session to work on.</param>
/// <returns>the MessageConsumer"/></returns>
/// <exception cref="EMSException">if thrown by EMS methods</exception>
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
/// <param name="session">The session to create a MessageConsumer for.</param>
/// <param name="destination">The destination to create a MessageConsumer for.</param>
/// <returns>The new MessageConsumer</returns>
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;

View File

@@ -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
/// </returns>
/// <throws>EMSException if thrown by EMS API methods </throws>
/// <throws>MessageConversionException in case of conversion failure </throws>
Message ToMessage(object objectToConvert, Session session);
Message ToMessage(object objectToConvert, ISession session);
/// <summary> Convert from a EMS Message to a .NET object.</summary>
/// <param name="messageToConvert">the message to convert

View File

@@ -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
/// </returns>
/// <throws>EMSException if thrown by EMS API methods </throws>
/// <throws>MessageConversionException in case of conversion failure </throws>
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
/// <returns> the resulting message
/// </returns>
/// <throws> EMSException if thrown by EMS methods </throws>
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
/// <returns> the resulting message
/// </returns>
/// <throws> EMSException if thrown by EMS methods </throws>
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
/// <returns> the resulting message
/// </returns>
/// <throws> EMSException if thrown by EMS methods </throws>
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
/// </returns>
/// <throws> EMSException if thrown by EMS methods </throws>
protected virtual ObjectMessage CreateMessageForSerializable(
object objectToSend, Session session)
object objectToSend, ISession session)
{
return session.CreateObjectMessage(objectToSend);
}

View File

@@ -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
/// <returns>the NMS Message</returns>
/// <throws>NMSException if thrown by NMS API methods </throws>
/// <throws>MessageConversionException in case of conversion failure </throws>
public Message ToMessage(object objectToConvert, Session session)
public Message ToMessage(object objectToConvert, ISession session)
{
if (objectToConvert == null)
{

View File

@@ -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
/// <returns> the EMS destination (either a topic or a queue)
/// </returns>
/// <throws>EMSException if resolution failed </throws>
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
/// <returns> the EMS Topic name
/// </returns>
/// <throws>EMSException if resolution failed </throws>
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
/// <returns> the EMS Queue name
/// </returns>
/// <throws>EMSException if resolution failed </throws>
protected internal virtual Destination ResolveQueue(Session session, string queueName)
protected internal virtual Destination ResolveQueue(ISession session, string queueName)
{
return session.CreateQueue(queueName);
}

View File

@@ -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
/// <param name="destinationName">Name of the destination.</param>
/// <returns>The located Destination</returns>
/// <exception cref="EMSException">If resolution failed.</exception>
public virtual Destination ResolveDestinationName(Session session, System.String destinationName)
public virtual Destination ResolveDestinationName(ISession session, System.String destinationName)
{
return DestinationResolver.ResolveDestinationName(session, destinationName, PubSubDomain);
}

View File

@@ -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
/// <returns> the EMS destination (either a topic or a queue)
/// </returns>
/// <throws>EMSException if resolution failed </throws>
Destination ResolveDestinationName(Session session, string destinationName, bool pubSubDomain);
Destination ResolveDestinationName(ISession session, string destinationName, bool pubSubDomain);
}
}

View File

@@ -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.
/// </summary>
/// <value>The connection factory.</value>
virtual public ConnectionFactory ConnectionFactory
virtual public IConnectionFactory ConnectionFactory
{
get
{
@@ -142,7 +143,7 @@ namespace Spring.Messaging.Ems.Support
/// Creates the connection via the ConnectionFactory.
/// </summary>
/// <returns></returns>
protected virtual Connection CreateConnection()
protected virtual IConnection CreateConnection()
{
return ConnectionFactory.CreateConnection();
}
@@ -152,7 +153,7 @@ namespace Spring.Messaging.Ems.Support
/// </summary>
/// <param name="con">The connection to create a session for.</param>
/// <returns>The new session</returns>
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
/// </summary>
/// <param name="session">The session to check.</param>
/// <returns>true if in client ack mode, false otherwise</returns>
protected virtual bool IsClientAcknowledge(Session session)
protected virtual bool IsClientAcknowledge(ISession session)
{
return (session.AcknowledgeMode == Session.CLIENT_ACKNOWLEDGE);
}

View File

@@ -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
/// </summary>
/// <param name="con">the EMS Connection to close (may be <code>null</code>)
/// </param>
public static void CloseConnection(Connection con)
public static void CloseConnection(IConnection con)
{
CloseConnection(con, false);
}
@@ -54,7 +55,7 @@ namespace Spring.Messaging.Ems.Support
/// </param>
/// <param name="stop">whether to call <code>stop()</code> before closing
/// </param>
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
/// </summary>
/// <param name="session">the EMS Session to close (may be <code>null</code>)
/// </param>
public static void CloseSession(Session session)
public static void CloseSession(ISession session)
{
if (session != null)
{
@@ -118,7 +119,7 @@ namespace Spring.Messaging.Ems.Support
/// </summary>
/// <param name="producer">the EMS MessageProducer to close (may be <code>null</code>)
/// </param>
public static void CloseMessageProducer(MessageProducer producer)
public static void CloseMessageProducer(IMessageProducer producer)
{
if (producer != null)
{
@@ -143,7 +144,7 @@ namespace Spring.Messaging.Ems.Support
/// </summary>
/// <param name="consumer">the EMS MessageConsumer to close (may be <code>null</code>)
/// </param>
public static void CloseMessageConsumer(MessageConsumer consumer)
public static void CloseMessageConsumer(IMessageConsumer consumer)
{
if (consumer != null)
{
@@ -195,7 +196,7 @@ namespace Spring.Messaging.Ems.Support
/// <param name="session">the EMS Session to commit
/// </param>
/// <throws>EMSException if committing failed </throws>
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
/// <param name="session">the EMS Session to rollback
/// </param>
/// <throws> EMSException if committing failed </throws>
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
}
}
}
/// <summary>
/// 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
/// </summary>
/// <param name="ackMode">The ack mode.</param>
/// <returns>The corresponding SessionMode enumeration</returns>
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;
}
}
}
}

View File

@@ -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);

View File

@@ -2,7 +2,10 @@
<objects xmlns="http://www.springframework.net">
<object id="ConnectionFactory" type="TIBCO.EMS.ConnectionFactory, TIBCO.EMS">
<object id="ConnectionFactory" type="Spring.Messaging.Ems.Common.EmsConnectionFactory, Spring.Messaging.Ems">
<constructor-arg ref="NativeConnectionFactory"/>
</object>
<object id="NativeConnectionFactory" type="TIBCO.EMS.ConnectionFactory, TIBCO.EMS">
<constructor-arg index="0" value="tcp://localhost:7222"/>
</object>

View File

@@ -45,6 +45,10 @@
<SpecificVersion>False</SpecificVersion>
<HintPath>..\..\..\lib\Net\2.0\TIBCO.EMS.dll</HintPath>
</Reference>
<Reference Include="TIBCO.EMS.ADMIN, Version=1.0.0.0, Culture=neutral, PublicKeyToken=5b83db8ff05c64ba, processorArchitecture=MSIL">
<SpecificVersion>False</SpecificVersion>
<HintPath>..\..\..\lib\Net\2.0\TIBCO.EMS.ADMIN.dll</HintPath>
</Reference>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\..\src\Spring\Spring.Aop\Spring.Aop.2005.csproj">