diff options
Diffstat (limited to 'qpid/wcf/src/Apache/Qpid/Interop/OutputLink.cpp')
| -rw-r--r-- | qpid/wcf/src/Apache/Qpid/Interop/OutputLink.cpp | 255 |
1 files changed, 255 insertions, 0 deletions
diff --git a/qpid/wcf/src/Apache/Qpid/Interop/OutputLink.cpp b/qpid/wcf/src/Apache/Qpid/Interop/OutputLink.cpp new file mode 100644 index 0000000000..de7141dadb --- /dev/null +++ b/qpid/wcf/src/Apache/Qpid/Interop/OutputLink.cpp @@ -0,0 +1,255 @@ +/* +* 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 "qpid/client/AsyncSession.h" +#include "qpid/framing/FrameSet.h" +#include "qpid/client/SubscriptionManager.h" +#include "qpid/client/Connection.h" +#include "qpid/client/Message.h" +#include "qpid/client/MessageListener.h" + + +#include "AmqpSession.h" +#include "AmqpMessage.h" +#include "OutputLink.h" +#include "QpidMarshal.h" + +namespace Apache { +namespace Qpid { +namespace Interop { + +using namespace System; +using namespace System::Runtime::InteropServices; +using namespace System::Threading; +using namespace msclr; + +using namespace qpid::client; +using namespace std; + +using namespace Apache::Qpid::AmqpTypes; + + +OutputLink::OutputLink(AmqpSession^ session, String^ address) : + amqpSession(session), + disposed(false), + maxFrameSize(session->Connection->MaxFrameSize), + finalizing(false) +{ + qpidAddress = QpidAddress::CreateAddress(address, false); + qpidAddress->ResolveLink(session); +} + +void OutputLink::Cleanup() +{ + { + lock l(this); + if (disposed) + return; + + disposed = true; + } + + // process any pending queue delete + qpidAddress->CleanupLink(amqpSession); + amqpSession->NotifyClosed(); +} + +OutputLink::~OutputLink() +{ + Cleanup(); +} + +OutputLink::!OutputLink() +{ + Cleanup(); +} + +void OutputLink::Close() +{ + // Simulate Dispose()... + Cleanup(); + GC::SuppressFinalize(this); +} + + +AmqpMessage^ OutputLink::CreateMessage() +{ + MessageBodyStream ^mbody = gcnew MessageBodyStream(maxFrameSize); + AmqpMessage ^amqpm = gcnew AmqpMessage(mbody); + return amqpm; +} + + +void OutputLink::ManagedToNative(AmqpMessage^ m) +{ + MessageBodyStream^ messageBodyStream = (MessageBodyStream^ ) m->BodyStream; + + AmqpProperties^ mprops = m->Properties; + + if (mprops != nullptr) { + AMQHeaderBody* bodyp = (AMQHeaderBody*) messageBodyStream->GetHeader().ToPointer(); + + if (mprops->HasDeliveryProperties) { + DeliveryProperties* deliveryPropertiesp = bodyp->get<DeliveryProperties>(true); + + if (mprops->RoutingKey != nullptr) { + deliveryPropertiesp->setRoutingKey(QpidMarshal::ToNative(mprops->RoutingKey)); + } + + if (mprops->Durable) { + deliveryPropertiesp->setDeliveryMode(qpid::framing::PERSISTENT); + } + + if (mprops->TimeToLive.HasValue) { + long long ttl = mprops->TimeToLive.Value.Ticks; + bool was_positive = (ttl > 0); + if (ttl < 0) + ttl = 0; + ttl = ttl / TimeSpan::TicksPerMillisecond; + if ((ttl == 0) && was_positive) + ttl = 1; + deliveryPropertiesp->setTtl(ttl); + } + } + + if (mprops->HasMessageProperties) { + qpid::framing::MessageProperties* messagePropertiesp = + bodyp->get<qpid::framing::MessageProperties>(true); + + String^ replyToExchange = mprops->ReplyToExchange; + String^ replyToRoutingKey = mprops->ReplyToRoutingKey; + if ((replyToExchange != nullptr) || (replyToRoutingKey != nullptr)) { + qpid::framing::ReplyTo nReplyTo; + if (replyToExchange != nullptr) { + nReplyTo.setExchange(QpidMarshal::ToNative(replyToExchange)); + } + if (replyToRoutingKey != nullptr) { + nReplyTo.setRoutingKey(QpidMarshal::ToNative(replyToRoutingKey)); + } + messagePropertiesp->setReplyTo(nReplyTo); + } + + // TODO: properly split 1.0 style to 0-10 content type + encoding + + String^ contentType = mprops->ContentType; + if (contentType != nullptr) { + String^ type = nullptr; + String^ enc = nullptr; + int idx = contentType->IndexOf(';'); + if (idx == -1) { + type = contentType; + } + else { + type = contentType->Substring(0, idx); + contentType = contentType->Substring(idx + 1); + idx = contentType->IndexOf('='); + if (idx != -1) { + enc = contentType->Substring(idx + 1); + enc = enc->Trim(); + } + } + if (!String::IsNullOrEmpty(type)) { + messagePropertiesp->setContentType(QpidMarshal::ToNative(type)); + } + if (!String::IsNullOrEmpty(enc)) { + messagePropertiesp->setContentEncoding(QpidMarshal::ToNative(enc)); + } + } + + + array<unsigned char>^ mbytes = mprops->CorrelationId; + if (mbytes != nullptr) { + pin_ptr<unsigned char> pinnedBuf = &mbytes[0]; + std::string s((char *) pinnedBuf, mbytes->Length); + messagePropertiesp->setCorrelationId(s); + } + + mbytes = mprops->UserId; + if (mbytes != nullptr) { + pin_ptr<unsigned char> pinnedBuf = &mbytes[0]; + std::string s((char *) pinnedBuf, mbytes->Length); + messagePropertiesp->setUserId(s); + } + + if (mprops->HasMappedProperties) { + qpid::framing::FieldTable fieldTable; + // TODO: add support for abitrary AMQP types + for each (Collections::Generic::KeyValuePair<System::String^, AmqpType^> kvp in mprops->PropertyMap) { + Type^ type = kvp.Value->GetType(); + if (type == AmqpInt::typeid) { + fieldTable.setInt(QpidMarshal::ToNative(kvp.Key), + ((AmqpInt ^) kvp.Value)->Value); + } + else if (type == AmqpString::typeid) { + AmqpString^ str = (AmqpString ^) kvp.Value; + // For now, FieldTable supports a single string type + fieldTable.setString(QpidMarshal::ToNative(kvp.Key), QpidMarshal::ToNative(str->Value)); + } + } + + messagePropertiesp->setApplicationHeaders(fieldTable); + } + } + } +} + + + +void OutputLink::Send(AmqpMessage^ amqpMessage, TimeSpan timeout) +{ + // copy properties from managed space to the native counterparts + ManagedToNative(amqpMessage); + + MessageBodyStream^ messageBodyStream = (MessageBodyStream^ ) amqpMessage->BodyStream; + CompletionWaiter^ waiter = amqpSession->SendMessage(qpidAddress->LinkName, messageBodyStream, + timeout, false, nullptr, nullptr); + + if (waiter != nullptr) { + waiter->WaitForCompletion(); + if (waiter->TimedOut) { + throw gcnew TimeoutException("Receive"); + } + } + // else: SendMessage() has already waited for the Completion + +} + +IAsyncResult^ OutputLink::BeginSend(AmqpMessage^ amqpMessage, TimeSpan timeout, AsyncCallback^ callback, Object^ state) +{ + ManagedToNative(amqpMessage); + + MessageBodyStream^ messageBodyStream = (MessageBodyStream^ ) amqpMessage->BodyStream; + CompletionWaiter^ waiter = amqpSession->SendMessage(qpidAddress->LinkName, messageBodyStream, timeout, true, callback, state); + return waiter; +} + +void OutputLink::EndSend(IAsyncResult^ result) +{ + CompletionWaiter^ waiter = (CompletionWaiter ^) result; + waiter->WaitForCompletion(); + if (waiter->TimedOut) { + throw gcnew TimeoutException("Receive"); + } +} + + +}}} // namespace Apache::Qpid::Interop |