Go to the documentation of this file.00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00022
00023
00024
00025
00026
00027
00028
00029
00030
00031
00032
00033
00034
00035
00036
00037
00038
00039 #include "../Port.hpp"
00040 #include "ConnFactory.hpp"
00041 #include "../base/InputPortInterface.hpp"
00042 #include "../DataFlowInterface.hpp"
00043 #include "../types/TypeMarshaller.hpp"
00044
00045 using namespace std;
00046 using namespace RTT;
00047 using namespace RTT::internal;
00048
00049 bool LocalConnID::isSameID(ConnID const& id) const
00050 {
00051 LocalConnID const* real_id = dynamic_cast<LocalConnID const*>(&id);
00052 if (!real_id)
00053 return false;
00054 else return real_id->ptr == this->ptr;
00055 }
00056
00057 ConnID* LocalConnID::clone() const {
00058 return new LocalConnID(this->ptr);
00059 }
00060
00061 bool StreamConnID::isSameID(ConnID const& id) const
00062 {
00063 StreamConnID const* real_id = dynamic_cast<StreamConnID const*>(&id);
00064 if (!real_id)
00065 return false;
00066 else return real_id->name_id == this->name_id;
00067 }
00068
00069 ConnID* StreamConnID::clone() const {
00070 return new StreamConnID(this->name_id);
00071 }
00072
00073 base::ChannelElementBase::shared_ptr RTT::internal::ConnFactory::createRemoteConnection(base::OutputPortInterface& output_port, base::InputPortInterface& input_port, const ConnPolicy& policy)
00074 {
00075
00076
00077
00078 int transport = policy.transport == 0 ? input_port.serverProtocol() : policy.transport;
00079 types::TypeInfo const* type_info = output_port.getTypeInfo();
00080 if (!type_info || input_port.getTypeInfo() != type_info)
00081 {
00082 log(Error) << "Type of port " << output_port.getName() << " is not registered into the type system, cannot marshal it into the right transporter" << endlog();
00083
00084 return base::ChannelElementBase::shared_ptr();
00085 }
00086 else if ( !type_info->getProtocol( transport ) )
00087 {
00088 log(Error) << "Type " << type_info->getTypeName() << " cannot be marshalled into the requested transporter (id:"<< transport<<")." << endlog();
00089
00090 return base::ChannelElementBase::shared_ptr();
00091 }
00092 else
00093 {
00094 return input_port.
00095 buildRemoteChannelOutput(output_port, type_info, input_port, policy);
00096 }
00097 return base::ChannelElementBase::shared_ptr();
00098 }
00099
00100 bool ConnFactory::createAndCheckConnection(base::OutputPortInterface& output_port, base::InputPortInterface& input_port, base::ChannelElementBase::shared_ptr channel_input, ConnPolicy policy) {
00101
00102 if ( output_port.addConnection( input_port.getPortID(), channel_input, policy ) ) {
00103
00104 if ( input_port.channelReady( channel_input->getOutputEndPoint(), policy ) == false ) {
00105 output_port.disconnect( &input_port );
00106 log(Error) << "The input port "<< input_port.getName()
00107 << " could not successfully read from the connection from output port " << output_port.getName() <<endlog();
00108
00109 return false;
00110 }
00111 log(Debug) << "Connected output port "<< output_port.getName()
00112 << " successfully to " << input_port.getName() <<endlog();
00113 return true;
00114 }
00115
00116 channel_input->disconnect(true);
00117 log(Error) << "The output port "<< output_port.getName()
00118 << " could not successfully use the connection to input port " << input_port.getName() <<endlog();
00119 return false;
00120 }
00121
00122 bool ConnFactory::createAndCheckStream(base::OutputPortInterface& output_port, ConnPolicy const& policy, base::ChannelElementBase::shared_ptr chan, StreamConnID* conn_id) {
00123 if (policy.transport == 0 ) {
00124 log(Error) << "Need a transport for creating streams." <<endlog();
00125 return false;
00126 }
00127 const types::TypeInfo* type = output_port.getTypeInfo();
00128 if ( type->getProtocol(policy.transport) == 0 ) {
00129 log(Error) << "Could not create transport stream for port "<< output_port.getName() << " with transport id " << policy.transport <<endlog();
00130 log(Error) << "No such transport registered. Check your policy.transport settings or add the transport for type "<< type->getTypeName() <<endlog();
00131 return false;
00132 }
00133 types::TypeMarshaller* ttt = dynamic_cast<types::TypeMarshaller*> ( type->getProtocol(policy.transport) );
00134 if (ttt) {
00135 int size_hint = ttt->getSampleSize( output_port.getDataSource() );
00136 policy.data_size = size_hint;
00137 } else {
00138 log(Debug) <<"Could not determine sample size for type " << type->getTypeName() << endlog();
00139 }
00140 RTT::base::ChannelElementBase::shared_ptr chan_stream = type->getProtocol(policy.transport)->createStream(&output_port, policy, true);
00141
00142 if ( !chan_stream ) {
00143 log(Error) << "Transport failed to create remote channel for output stream of port "<<output_port.getName() << endlog();
00144 return false;
00145 }
00146 chan->setOutput( chan_stream );
00147
00148 if ( output_port.addConnection( new StreamConnID(policy.name_id), chan, policy) ) {
00149 log(Info) << "Created output stream for output port "<< output_port.getName() <<endlog();
00150 return true;
00151 }
00152
00153 log(Error) << "Failed to create output stream for output port "<< output_port.getName() <<endlog();
00154 return false;
00155 }
00156
00157 bool ConnFactory::createAndCheckStream(base::InputPortInterface& input_port, ConnPolicy const& policy, base::ChannelElementBase::shared_ptr outhalf, StreamConnID* conn_id) {
00158 if (policy.transport == 0 ) {
00159 log(Error) << "Need a transport for creating streams." <<endlog();
00160 return false;
00161 }
00162 const types::TypeInfo* type = input_port.getTypeInfo();
00163 if ( type->getProtocol(policy.transport) == 0 ) {
00164 log(Error) << "Could not create transport stream for port "<< input_port.getName() << " with transport id " << policy.transport <<endlog();
00165 log(Error) << "No such transport registered. Check your policy.transport settings or add the transport for type "<< type->getTypeName() <<endlog();
00166 return false;
00167 }
00168
00169
00170
00171 RTT::base::ChannelElementBase::shared_ptr chan = type->getProtocol(policy.transport)->createStream(&input_port,policy, false);
00172
00173 if ( !chan ) {
00174 log(Error) << "Transport failed to create remote channel for input stream of port "<<input_port.getName() << endlog();
00175 return false;
00176 }
00177
00178 conn_id->name_id = policy.name_id;
00179
00180 chan->getOutputEndPoint()->setOutput( outhalf );
00181 if ( input_port.channelReady( chan->getOutputEndPoint(), policy ) == true ) {
00182 log(Info) << "Created input stream for input port "<< input_port.getName() <<endlog();
00183 return true;
00184 }
00185
00186 chan = 0;
00187 log(Error) << "Failed to create input stream for input port "<< input_port.getName() <<endlog();
00188 return false;
00189 }
00190
00191 base::ChannelElementBase::shared_ptr ConnFactory::createAndCheckOutOfBandConnection( base::OutputPortInterface& output_port,
00192 base::InputPortInterface& input_port,
00193 ConnPolicy const& policy,
00194 base::ChannelElementBase::shared_ptr output_half,
00195 StreamConnID* conn_id)
00196 {
00197
00198 const types::TypeInfo* type = output_port.getTypeInfo();
00199 if ( type->getProtocol(policy.transport) == 0 ) {
00200 log(Error) << "Could not create out-of-band transport for port "<< output_port.getName() << " with transport id " << policy.transport <<endlog();
00201 log(Error) << "No such transport registered. Check your policy.transport settings or add the transport for type "<< type->getTypeName() <<endlog();
00202 return 0;
00203 }
00204
00205
00206 ConnPolicy policy2 = policy;
00207 policy2.pull = false;
00208 conn_id->name_id = policy2.name_id;
00209
00210
00211 types::TypeMarshaller* ttt = dynamic_cast<types::TypeMarshaller*>( type->getProtocol(policy.transport) );
00212 if (ttt) {
00213 policy2.data_size = ttt->getSampleSize( output_port.getDataSource() );
00214 } else {
00215 log(Debug) <<"Could not determine sample size for type " << type->getTypeName() << endlog();
00216 }
00217
00218 if ( input_port.isLocal() ) {
00219 RTT::base::ChannelElementBase::shared_ptr ceb_input = type->getProtocol(policy.transport)->createStream(&input_port, policy2, false);
00220 if (ceb_input) {
00221 log(Info) <<"Receiving data for port "<<input_port.getName() << " from out-of-band protocol "<< policy.transport << " with id "<< policy2.name_id<<endlog();
00222 } else {
00223 log(Error) << "The type transporter for type "<<type->getTypeName()<< " failed to create a remote channel for port " << input_port.getName()<<endlog();
00224 return 0;
00225 }
00226 ceb_input->getOutputEndPoint()->setOutput(output_half);
00227 output_half = ceb_input;
00228 }
00229
00230
00231 if ( output_port.isLocal() ) {
00232
00233 RTT::base::ChannelElementBase::shared_ptr ceb_output = type->getProtocol(policy.transport)->createStream(&output_port, policy2, true);
00234 if (ceb_output) {
00235 log(Info) <<"Redirecting data for port "<< output_port.getName() << " to out-of-band protocol "<< policy.transport << " with id "<< policy2.name_id <<endlog();
00236 } else {
00237 log(Error) << "The type transporter for type "<<type->getTypeName()<< " failed to create a remote channel for port " << output_port.getName()<<endlog();
00238 return 0;
00239 }
00240
00241
00242 ceb_output->getOutputEndPoint()->setOutput(output_half);
00243 output_half = ceb_output;
00244 }
00245
00246 policy.name_id = policy2.name_id;
00247 conn_id->name_id = policy2.name_id;
00248
00249 return output_half;
00250
00251 }
00252