summaryrefslogtreecommitdiff
path: root/qpid/wcf/src/Apache/Qpid/Interop/AmqpConnection.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'qpid/wcf/src/Apache/Qpid/Interop/AmqpConnection.cpp')
-rw-r--r--qpid/wcf/src/Apache/Qpid/Interop/AmqpConnection.cpp276
1 files changed, 276 insertions, 0 deletions
diff --git a/qpid/wcf/src/Apache/Qpid/Interop/AmqpConnection.cpp b/qpid/wcf/src/Apache/Qpid/Interop/AmqpConnection.cpp
new file mode 100644
index 0000000000..1bc9a15d92
--- /dev/null
+++ b/qpid/wcf/src/Apache/Qpid/Interop/AmqpConnection.cpp
@@ -0,0 +1,276 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+
+#include <windows.h>
+#include <msclr\lock.h>
+#include <oletx2xa.h>
+
+#include "qpid/client/AsyncSession.h"
+#include "qpid/client/SubscriptionManager.h"
+#include "qpid/client/Connection.h"
+#include "qpid/client/Message.h"
+#include "qpid/client/MessageListener.h"
+#include "qpid/framing/FrameSet.h"
+
+#include "AmqpConnection.h"
+#include "AmqpSession.h"
+#include "QpidMarshal.h"
+#include "QpidException.h"
+#include "DtxResourceManager.h"
+#include "XaTransaction.h"
+
+namespace Apache {
+namespace Qpid {
+namespace Interop {
+
+using namespace System;
+using namespace System::Runtime::InteropServices;
+using namespace msclr;
+
+using namespace qpid::client;
+using namespace std;
+
+
+// Note on locks: Use thisLock for fast counting and idle/busy
+// notifications. Use the "sessions" list to serialize session
+// creation/reaping and overall tear down.
+
+
+AmqpConnection::AmqpConnection(String^ server, int port) :
+ connectionp(NULL),
+ busyCount(0),
+ disposed(false)
+{
+ initialize (server, port, false, false, nullptr, nullptr);
+}
+
+AmqpConnection::AmqpConnection(System::String^ server, int port, bool ssl, bool saslPlain, System::String^ username, System::String^ password) :
+ connectionp(NULL),
+ busyCount(0),
+ disposed(false)
+{
+ initialize (server, port, ssl, saslPlain, username, password);
+}
+
+void AmqpConnection::initialize(System::String^ server, int port, bool ssl, bool saslPlain, System::String^ username, System::String^ password)
+{
+ if (server == nullptr)
+ throw gcnew ArgumentNullException("AMQP server");
+ if (saslPlain) {
+ if (username == nullptr)
+ throw gcnew ArgumentNullException("username");
+ if (username == nullptr)
+ throw gcnew ArgumentNullException("password");
+ }
+
+ bool success = false;
+ System::Exception^ openException = nullptr;
+ sessions = gcnew Collections::Generic::List<AmqpSession^>();
+ thisLock = gcnew Object();
+
+ try {
+ connectionp = new Connection;
+
+ if (ssl || saslPlain) {
+ ConnectionSettings proposedSettings;
+ proposedSettings.host = QpidMarshal::ToNative(server);
+ proposedSettings.port = port;
+ if (ssl)
+ proposedSettings.protocol = "ssl";
+
+ if (saslPlain) {
+ proposedSettings.username = QpidMarshal::ToNative(username);
+ proposedSettings.password = QpidMarshal::ToNative(password);
+ proposedSettings.mechanism = "PLAIN";
+ }
+
+ connectionp->open (proposedSettings);
+ }
+ else {
+ connectionp->open (QpidMarshal::ToNative(server), port);
+ }
+
+ // TODO: registerFailureCallback for failover
+ success = true;
+ const ConnectionSettings& settings = connectionp->getNegotiatedSettings();
+ this->maxFrameSize = settings.maxFrameSize;
+ this->host = server;
+ this->port = port;
+ this->ssl = ssl;
+ this->saslPlain = saslPlain;
+ this->username = username;
+ this->password = password;
+ this->isOpen = true;
+ } catch (const qpid::Exception& error) {
+ String^ errmsg = gcnew String(error.what());
+ openException = gcnew QpidException(errmsg);
+ } finally {
+ if (!success) {
+ Cleanup();
+ if (openException == nullptr) {
+ openException = gcnew QpidException ("unknown connection failure");
+ }
+ throw openException;
+ }
+ }
+}
+
+AmqpConnection^ AmqpConnection::Clone() {
+ if (disposed)
+ throw gcnew ObjectDisposedException("AmqpConnection.Clone");
+ return gcnew AmqpConnection (this->host, this->port, this->ssl, this->saslPlain, this->username, this->password);
+}
+
+void AmqpConnection::Cleanup()
+{
+ {
+ lock l(sessions);
+ if (disposed)
+ return;
+ disposed = true;
+ }
+
+ try {
+ // let the child sessions clean up
+
+ for each(AmqpSession^ s in sessions) {
+ s->ConnectionClosed();
+ }
+ }
+ finally
+ {
+ if (connectionp != NULL) {
+ isOpen = false;
+ connectionp->close();
+ delete connectionp;
+ connectionp = NULL;
+ }
+ }
+}
+
+AmqpConnection::~AmqpConnection()
+{
+ Cleanup();
+}
+
+AmqpConnection::!AmqpConnection()
+{
+ Cleanup();
+}
+
+void AmqpConnection::Close()
+{
+ // Simulate Dispose()...
+ Cleanup();
+ GC::SuppressFinalize(this);
+}
+
+AmqpSession^ AmqpConnection::CreateSession()
+{
+ lock l(sessions);
+ if (disposed) {
+ throw gcnew ObjectDisposedException("AmqpConnection");
+ }
+ AmqpSession^ session = gcnew AmqpSession(this, connectionp);
+ sessions->Add(session);
+ return session;
+}
+
+// called whenever a child session becomes newly busy (a first reader or writer since last idle)
+
+void AmqpConnection::NotifyBusy()
+{
+ bool changed = false;
+ {
+ lock l(thisLock);
+ if (busyCount++ == 0)
+ changed = true;
+ }
+}
+
+// called whenever a child session becomes newly idle (a last reader or writer has closed)
+// The connection is idle when none of its child sessions are busy
+
+void AmqpConnection::NotifyIdle()
+{
+ bool connectionIdle = false;
+ {
+ lock l(thisLock);
+ if (--busyCount == 0)
+ connectionIdle = true;
+ }
+ if (connectionIdle) {
+ OnConnectionIdle(this, System::EventArgs::Empty);
+ }
+}
+
+void HexAppend(StringBuilder^ sb, String^ s) {
+ if (s->Length > 0) {
+ array<unsigned char>^ bytes = Encoding::UTF8->GetBytes(s);
+ for each (unsigned char b in bytes) {
+ sb->Append(String::Format("{0:x2}", b));
+ }
+ }
+ sb->Append(".");
+}
+
+
+// Note: any change to this format has to be reflected in the DTC plugin's xa_open()
+// for now: "QPIDdsnV2.port.host.instance_id.SSL_tf.SASL_mech.username.password"
+// This extended info is needed so that the DTC can make a separate connection to the broker
+// for recovery.
+
+String^ AmqpConnection::DataSourceName::get() {
+ if (dataSourceName == nullptr) {
+ StringBuilder^ sb = gcnew StringBuilder();
+ sb->Append("QPIDdsnV2.");
+
+ sb->Append(this->port);
+ sb->Append(".");
+
+ HexAppend(sb, this->host);
+
+ sb->Append(System::Diagnostics::Process::GetCurrentProcess()->Id);
+ sb->Append("-");
+ sb->Append(AppDomain::CurrentDomain->Id);
+ sb->Append(".");
+
+ if (this->ssl)
+ sb->Append("T");
+ else
+ sb->Append("F");
+ sb->Append(".");
+
+ if (this->saslPlain) {
+ sb->Append("P.");
+ HexAppend(sb, this->username);
+ HexAppend(sb, this->password);
+ }
+ else {
+ // SASL anonymous
+ sb->Append("A.");
+ }
+
+ dataSourceName = sb->ToString();
+ }
+ return dataSourceName;
+}
+
+
+}}} // namespace Apache::Qpid::Interop
View Lorry Controller Status