container_impl_base.hpp
#ifndef PROTON_IO_CONTAINER_IMPL_BASE_HPP
#define PROTON_IO_CONTAINER_IMPL_BASE_HPP
/*
*
* 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 "../container.hpp"
#include <mutex>
#include <sstream>
namespace proton {
namespace io {
class container_impl_base : public standard_container {
public:
// Pull in base class functions here so that name search finds all the overloads
using standard_container::open_receiver;
using standard_container::open_sender;
void client_connection_options(const connection_options & opts) {
store(client_copts_, opts);
}
connection_options client_connection_options() const {
return load(client_copts_);
}
void server_connection_options(const connection_options & opts) {
store(server_copts_, opts);
}
connection_options server_connection_options() const {
return load(server_copts_);
}
void sender_options(const class sender_options & opts) {
store(sender_opts_, opts);
}
class sender_options sender_options() const {
return load(sender_opts_);
}
void receiver_options(const class receiver_options & opts) {
store(receiver_opts_, opts);
}
class receiver_options receiver_options() const {
return load(receiver_opts_);
}
returned<sender> open_sender(
const std::string &url, const class sender_options &opts, const connection_options &copts)
{
return open_link<sender, class sender_options>(url, opts, copts, &connection::open_sender);
}
returned<receiver> open_receiver(
const std::string &url, const class receiver_options &opts, const connection_options &copts)
{
return open_link<receiver>(url, opts, copts, &connection::open_receiver);
}
private:
template<class T, class Opts>
returned<T> open_link(
const std::string &url_str, const Opts& opts, const connection_options& copts,
T (connection::*open_fn)(const std::string&, const Opts&))
{
std::string addr = url(url_str).path();
std::shared_ptr<thread_safe<connection> > ts_connection = connect(url_str, copts);
std::promise<returned<T> > result_promise;
auto do_open = [ts_connection, addr, opts, open_fn, &result_promise]() {
try {
connection c = ts_connection->unsafe();
returned<T> s = make_thread_safe((c.*open_fn)(addr, opts));
result_promise.set_value(s);
} catch (...) {
result_promise.set_exception(std::current_exception());
}
};
ts_connection->event_loop()->inject(do_open);
std::future<returned<T> > result_future = result_promise.get_future();
if (!result_future.valid())
throw error(url_str+": connection closed");
return result_future.get();
}
mutable std::mutex lock_;
template <class T> T load(const T& v) const {
std::lock_guard<std::mutex> g(lock_);
return v;
}
template <class T> void store(T& v, const T& x) const {
std::lock_guard<std::mutex> g(lock_);
v = x;
}
connection_options client_copts_, server_copts_;
class receiver_options receiver_opts_;
class sender_options sender_opts_;
};
} // io
} // proton
#endif // PROTON_IO_CONTAINER_IMPL_BASE_HPP
proton::connection::open_receiver
receiver open_receiver(const std::string &addr)
Open a receiver for addr on default_session().
proton::sender_options
Options for creating a sender.
Definition: sender_options.hpp:64
proton::io::container_impl_base::client_connection_options
void client_connection_options(const connection_options &opts)
Connection options that will be to outgoing connections.
Definition: container_impl_base.hpp:48
proton::connection
A connection to a remote AMQP peer.
Definition: connection.hpp:48
proton::connection_options
Options for creating a connection.
Definition: connection_options.hpp:67
proton::url::path
std::string path() const
path is everything after the final "/".
proton::url
A Proton URL.
Definition: url.hpp:55
proton::io::container_impl_base
Experimental - A base container implementation.
Definition: container_impl_base.hpp:41
proton::io::container_impl_base::client_connection_options
connection_options client_connection_options() const
Connection options that will be to outgoing connections.
Definition: container_impl_base.hpp:53
proton::io::container_impl_base::open_sender
returned< sender > open_sender(const std::string &url, const class sender_options &opts, const connection_options &copts)
Open a connection and sender for url.
Definition: container_impl_base.hpp:88
proton::io::container_impl_base::server_connection_options
connection_options server_connection_options() const
Connection options that will be applied to incoming connections.
Definition: container_impl_base.hpp:63
proton::receiver_options
Options for creating a receiver.
Definition: receiver_options.hpp:62
proton::io::container_impl_base::sender_options
void sender_options(const class sender_options &opts)
Sender options applied to senders created by this container.
Definition: container_impl_base.hpp:68
proton::connection::open_sender
sender open_sender(const std::string &addr)
Open a sender for addr on default_session().
proton::io::container_impl_base::receiver_options
void receiver_options(const class receiver_options &opts)
Receiver options applied to receivers created by this container.
Definition: container_impl_base.hpp:78
proton::io::container_impl_base::open_receiver
returned< receiver > open_receiver(const std::string &url, const class receiver_options &opts, const connection_options &copts)
Open a connection and receiver for url.
Definition: container_impl_base.hpp:95
proton::io::container_impl_base::server_connection_options
void server_connection_options(const connection_options &opts)
Connection options that will be applied to incoming connections.
Definition: container_impl_base.hpp:58
proton::error
The base Proton error.
Definition: error.hpp:37
proton::make_thread_safe
returned< T > make_thread_safe(const T &obj)
Make a thread-safe wrapper for obj.
Definition: thread_safe.hpp:165
proton
Type traits for mapping between AMQP and C++ types.
Definition: annotation_key.hpp:28