31 f_running_callback( [](){
return; } )
41 string_t t_name = p_node->get_name();
45 msg_normal( coremsg,
"initializing node <" << t_name <<
">" <<
eom );
50 catch( std::exception& e )
52 msg_error( coremsg,
"exception caught while initializing node <" << t_name <<
">: " << e.what() <<
eom );
57 msg_normal( coremsg,
"added node <" << t_name <<
">" <<
eom );
60 if( t_inst !=
nullptr )
67 throw error() <<
"root add found preexisting node with name <" << t_name <<
">";
76 t_pos >= p_string.length()-2 ||
79 throw error() <<
"connection specification <" << p_string <<
"> was not formatted correctly: signal_node.signal:slot_node.slot";
83 string_t t_signal_argument = p_string.substr( 0, t_pos );
88 t_pos == string::npos ||
91 throw error() <<
"signal specification <" << t_signal_argument <<
"> is not formatted correctly: signal_node.signal";
95 string_t t_signal_node_string = t_signal_argument.substr( 0, t_pos );
96 string_t t_signal_string = t_signal_argument.substr( t_pos +
s_designator.length(), string_t::npos );
100 t_pos == string::npos ||
103 throw error() <<
"slot specification <" << t_slot_argument <<
"> is not formatted correctly: slot_node.slot";
107 string_t t_slot_node_string = t_slot_argument.substr( 0, t_pos );
108 string_t t_slot_string = t_slot_argument.substr( t_pos +
s_designator.length(), string_t::npos );
111 if( t_node_it ==
f_nodes.end() )
113 throw error() <<
"signal node was not found <" << t_signal_node_string <<
">";
116 node* t_signal_node = t_node_it->second;
118 t_node_it =
f_nodes.find( t_slot_node_string );
119 if( t_node_it ==
f_nodes.end() )
121 throw error() <<
"slot node was not found <" << t_slot_node_string <<
">";
124 node* t_slot_node = t_node_it->second;
127 if( t_signal == NULL )
129 throw error() <<
"signal <" << t_signal_string <<
"> was not found for node <" << t_signal_node <<
">";
136 throw error() <<
"slot <" << t_slot_string <<
"> was not found for node <" << t_slot_node_string <<
">";
142 msg_normal( coremsg,
"connected signal --> slot: " << t_signal_node_string <<
"." << t_signal_string <<
" --> " << t_slot_node_string <<
"." << t_slot_string <<
eom );
164 if( t_first_pos != string_t::npos )
167 if( t_second_pos == string_t::npos )
169 t_first_argument = p_string.substr( 0, t_first_pos );
170 t_second_argument = p_string.substr( t_first_pos + 1, string_t::npos );
173 if( t_first_pos != string_t::npos )
176 if( t_second_pos == string_t::npos )
178 t_first_node_string = t_first_argument.substr( 0, t_first_pos );
179 t_first_out_string = t_first_argument.substr( t_first_pos + 1, string_t::npos );
183 throw error() <<
"root join found multiple designators in first argument <" << t_first_argument <<
">";
189 if( t_first_pos != string_t::npos )
192 if( t_second_pos == string_t::npos )
194 t_second_node_string = t_second_argument.substr( 0, t_first_pos );
195 t_second_in_string = t_second_argument.substr( t_first_pos + 1, string_t::npos );
199 throw error() <<
"root join found multiple designators in second argument <" << t_second_argument <<
">";
204 if( t_first_out_string.empty() != t_second_in_string.empty() )
206 throw error() <<
"root join must link either by stream or pointer";
211 if( t_first_it ==
f_nodes.end() )
213 throw error() <<
"root join found no first node with name <" << t_first_node_string <<
">";
216 t_first_node = t_first_it->second;
219 if( t_second_it ==
f_nodes.end() )
221 throw error() <<
"root join found no second node with name <" << t_second_node_string <<
">";
224 t_second_node = t_second_it->second;
226 if( t_first_out_string.empty() )
229 t_first_node->
node_ptr( t_second_node, t_second_node_string );
231 msg_normal( coremsg,
"joined <" << t_first_node_string <<
"> with <" << t_second_node_string <<
">" <<
eom );
236 t_first_out = t_first_node->
out( t_first_out_string );
237 if( t_first_out == NULL )
239 throw error() <<
"root join found no first out with name <" << t_first_out_string <<
"> in node with name <" << t_first_node_string <<
">";
243 t_second_in = t_second_node->
in( t_second_in_string );
244 if( t_second_in == NULL )
246 throw error() <<
"root join found no second in with name <" << t_second_in_string <<
"> in node with name <" << t_second_node_string <<
">";
249 t_second_in->set( t_first_out->get() );
251 msg_normal( coremsg,
"joined <" << t_first_node_string <<
"." << t_first_out_string <<
"> with <" << t_second_node_string <<
"." << t_second_in_string <<
">" <<
eom );
258 throw error() <<
"root join found multiple connectors in string_t <" << p_string <<
">";
264 throw error() <<
"root join found no connector in string_t <" << p_string <<
">";
272 size_t t_separator_pos;
279 t_argument = p_string;
287 if( t_inst !=
nullptr )
296 t_separator_pos = t_argument.find(
s_separator, t_start_pos );
298 t_node_name = t_argument.substr( t_start_pos, t_separator_pos - t_start_pos );
299 t_argument = t_argument.substr( t_separator_pos +
s_separator.size(), string_t::npos );
301 if( t_node_name.size() == 0 )
303 return std::make_exception_ptr(
error() <<
"root run found node name with length zero in argument <" << p_string <<
">" );
306 t_node_it =
f_nodes.find( t_node_name );
307 if( t_node_it ==
f_nodes.end() )
309 return std::make_exception_ptr(
error() <<
"root run found no node with name <" << t_node_name <<
">" );
312 msg_normal( coremsg,
"creating thread for node <" << t_node_name <<
">" <<
eom );
313 t_node = t_node_it->second;
316 if( t_separator_pos == string_t::npos )
323 std::this_thread::sleep_for( std::chrono::milliseconds( 100 ) );
327 t_threads_lock.unlock();
329 msg_normal( coremsg,
"waiting for threads to finish..." <<
eom );
335 t_threads_lock.lock();
353 std::rethrow_exception( e_ptr );
358 msg_error( coremsg,
"midge error thrown: " << e.what() <<
eom );
362 msg_error( coremsg,
"fatal error thrown: " << e.what() <<
eom );
366 msg_error( coremsg,
"non-fatal error thrown: " << e.what() <<
eom );
368 catch(
const std::exception& e )
370 msg_error( coremsg,
"non-node exception thrown: " << e.what() <<
eom );
372 msg_debug( coremsg,
"canceling run and setting exception pointer" <<
eom );
387 t_node = t_it->second;
393 t_node = t_it->second;
409 (*t_it)->instruct( p_inst );
420 msg_debug( coremsg,
"Canceling nodes: producers" <<
eom );
423 if( dynamic_cast< producer* >( t_it->second ) != nullptr )
425 msg_debug( coremsg,
"Canceling " << t_it->second->get_name() <<
eom );
426 t_it->second->cancel( a_code );
432 std::this_thread::sleep_for( std::chrono::milliseconds( 500 ) );
435 msg_debug( coremsg,
"Canceling nodes: transformers" <<
eom );
438 if( dynamic_cast< transformer* >( t_it->second ) != nullptr )
440 msg_debug( coremsg,
"Canceling " << t_it->second->get_name() <<
eom );
441 t_it->second->cancel();
445 msg_debug( coremsg,
"Canceling nodes: consumers" <<
eom );
448 if( dynamic_cast< consumer* >( t_it->second ) != nullptr )
450 msg_debug( coremsg,
"Canceling " << t_it->second->get_name() <<
eom );
451 t_it->second->cancel();
455 msg_debug( coremsg,
"Canceling nodes: bystanders" <<
eom );
458 if( dynamic_cast< bystander* >( t_it->second ) != nullptr )
460 msg_debug( coremsg,
"Canceling " << t_it->second->get_name() <<
eom );
461 t_it->second->cancel();
473 t_it->second->reset_cancel();
static const std::string & connector()
static const message_end eom
virtual void do_cancellation(int a_code)
void throw_ex(std::exception_ptr e_ptr)
To be used by running nodes to throw an exception.
static const std::string & designator()
inst_set_t::iterator inst_it_t
void instruct(instruction p_inst)
signal * signal_ptr(const std::string &p_label)
input * in(const std::string &p_label)
virtual void execute(diptera *)=0
virtual unsigned connect(slot *p_slot)=0
node * node_ptr(const std::string &p_label)
node_map_t::iterator node_it_t
static const std::string s_designator
thread_vector_t f_threads
static const std::string s_separator
void connect(const std::string &p_string)
Connect a signal to a slot: signal_node.signal:slot_node.slot.
virtual void do_reset_cancellation()
thread_vector_t::iterator thread_it_t
std::function< void() > f_running_callback
node_map_t::const_iterator node_cit_t
std::exception_ptr f_run_e_ptr
static const std::string & separator()
#define msg_debug(x_name, x_content)
static const std::string s_connector
node_map_t::value_type node_entry_t
std::mutex f_threads_mutex
output * out(const std::string &p_label)
void join(const std::string &p_string)
Join one node to another.
std::exception_ptr run(const std::string &p_string)
void add(node *p_node)
Add a node.
virtual void initialize()=0
inst_set_t f_instructables
#define msg_normal(x_name, x_content)
slot * slot_ptr(const std::string &p_label)
virtual void finalize()=0
#define msg_error(x_name, x_content)