Fawkes API
Fawkes Development Version
|
23 #include <core/exceptions/system.h>
24 #include <core/threading/mutex.h>
25 #include <core/threading/mutex_locker.h>
26 #include <core/threading/refc_rwlock.h>
27 #include <interface/interface.h>
28 #include <interface/mediators/interface_mediator.h>
29 #include <interface/mediators/message_mediator.h>
30 #include <utils/misc/strndup.h>
31 #include <utils/time/clock.h>
32 #include <utils/time/time.h>
57 : Exception(
"This interface instance '%s' of type '%s' is not opened for writing. %s",
75 :
Exception(
"This interface instance '%s' of type '%s' IS opened for writing, but "
76 "messages can only be enqueued on reading interfaces.",
94 :
Exception(
"Message of type '%s' cannot be enqueued in interface of type '%s'",
111 :
Exception(
"The interface %s (instance serial %u) is invalid. You cannot call %s anymore.",
235 write_access_ =
false;
238 next_message_id_ = 0;
240 fieldinfo_list_ = NULL;
241 messageinfo_list_ = NULL;
243 timestamp_ =
new Time(0, 0);
244 local_read_timestamp_ =
new Time(0, 0);
245 auto_timestamping_ =
true;
246 owner_ = strdup(
"?");
248 memset(hash_, 0, INTERFACE_HASH_SIZE_);
249 memset(hash_printable_, 0, INTERFACE_HASH_SIZE_ * 2 + 1);
258 data_mutex_ =
new Mutex();
267 delete message_queue_;
273 fieldinfo_list_ = fieldinfo_list_->
next;
275 finfol = fieldinfo_list_;
278 interface_messageinfo_t *minfol = messageinfo_list_;
280 messageinfo_list_ = messageinfo_list_->
next;
282 minfol = messageinfo_list_;
285 delete local_read_timestamp_;
297 const unsigned char *
309 return hash_printable_;
318 memcpy(hash_, ihash, INTERFACE_HASH_SIZE_);
319 for (
size_t s = 0; s < INTERFACE_HASH_SIZE_; ++s) {
320 snprintf(&hash_printable_[s * 2], 3,
"%02X", hash_[s]);
340 const char * enumtype,
348 newinfo->
name = name;
350 newinfo->
value = value;
352 newinfo->
next = NULL;
356 fieldinfo_list_ = newinfo;
359 while (infol->
next != NULL) {
362 infol->
next = newinfo;
377 interface_messageinfo_t *infol = messageinfo_list_;
378 interface_messageinfo_t *newinfo =
379 (interface_messageinfo_t *)malloc(
sizeof(interface_messageinfo_t));
382 newinfo->next = NULL;
386 messageinfo_list_ = newinfo;
389 while (infol->next != NULL) {
392 infol->
next = newinfo;
400 std::list<const char *>
403 std::list<const char *> types;
404 interface_messageinfo_t *cur = messageinfo_list_;
406 while (cur != NULL) {
407 types.push_back(cur->type);
421 return INTERFACE_HASH_SIZE_;
440 return write_access_;
478 *local_read_timestamp_ = *timestamp_;
496 if (!write_access_) {
497 throw InterfaceWriteDeniedException(type_, id_,
"Cannot write.");
504 if (auto_timestamping_)
506 long sec = 0, usec = 0;
539 Interface::set_type_id(
const char *type,
const char *
id)
541 strncpy(type_,
type, INTERFACE_TYPE_SIZE_);
542 strncpy(id_,
id, INTERFACE_ID_SIZE_);
543 snprintf(uid_, INTERFACE_UID_SIZE_ + 1,
"%s::%s", type_, id_);
546 type_[INTERFACE_TYPE_SIZE_] = 0;
547 id_[INTERFACE_ID_SIZE_] = 0;
548 uid_[INTERFACE_UID_SIZE_] = 0;
555 Interface::set_instance_serial(
unsigned short instance_serial)
557 instance_serial_ = instance_serial;
565 Interface::set_mediators(InterfaceMediator *iface_mediator, MessageMediator *msg_mediator)
567 interface_mediator_ = iface_mediator;
568 message_mediator_ = msg_mediator;
577 Interface::set_memory(
unsigned int serial,
void *real_ptr,
void *data_ptr)
580 mem_real_ptr_ = real_ptr;
589 Interface::set_readwrite(
bool write_access, RefCountRWLock *rwlock)
591 write_access_ = write_access;
599 Interface::set_owner(
const char *owner)
605 owner_ = strdup(
owner);
622 return ((strncmp(type_, comp.type_,
sizeof(type_)) == 0)
623 && (strncmp(id_, comp.id_,
sizeof(id_)) == 0));
633 return (strncmp(this->type_, interface_type,
sizeof(this->type_)) == 0);
685 return instance_serial_;
714 if (auto_timestamping_)
715 throw Exception(
"Auto timestamping enabled, cannot "
716 "set explicit timestamp");
718 throw Exception(
"Timestamp can only be set on writing "
745 auto_timestamping_ = enabled;
782 return (*timestamp_ != local_read_timestamp_);
854 return interface_mediator_->
writer(
this);
860 std::list<std::string>
863 return interface_mediator_->
readers(
this);
886 message->set_interface(
this);
887 message->
set_id(next_msg_id());
889 message_mediator_->
transmit(message);
890 unsigned int msgid = message->
id();
920 throw InterfaceMessageEnqueueException(type_, id_);
922 if (message == NULL) {
928 mcopy->set_interface(
this);
929 mcopy->
set_id(next_msg_id());
931 unsigned int msgid = mcopy->
id();
933 message->set_id(msgid);
936 throw InterfaceInvalidMessageException(
this, message);
952 if (!write_access_) {
953 throw InterfaceWriteDeniedException(type_,
955 "Cannot work on message queue on "
956 "reading instance of an interface (append).");
960 message_queue_->
append(message);
976 if (!write_access_) {
977 throw InterfaceWriteDeniedException(type_,
979 "Cannot work on message queue on "
980 "reading instance of an interface (remove msg).");
983 return message_queue_->
remove(message);
994 if (!write_access_) {
995 throw InterfaceWriteDeniedException(type_,
997 "Cannot work on message queue on "
998 "reading instance of an interface (remove id).");
1001 return message_queue_->
remove(message_id);
1011 if (!write_access_) {
1012 throw InterfaceWriteDeniedException(type_,
1014 "Cannot work on message queue on "
1015 "reading instance of an interface (size).");
1018 return message_queue_->
size();
1028 if (!write_access_) {
1029 throw InterfaceWriteDeniedException(type_,
1031 "Cannot work on message queue on "
1032 "reading instance of an interface (empty).");
1035 return message_queue_->
empty();
1045 if (!write_access_) {
1046 throw InterfaceWriteDeniedException(type_,
1048 "Cannot work on message queue on "
1049 "reading instance of an interface (flush).");
1052 message_queue_->
flush();
1064 if (!write_access_) {
1065 throw InterfaceWriteDeniedException(type_,
1067 "Cannot work on message queue on "
1068 "reading instance of an interface (lock).");
1071 message_queue_->
lock();
1085 if (!write_access_) {
1086 throw InterfaceWriteDeniedException(type_,
1088 "Cannot work on message queue on "
1089 "reading instance of an interface "
1090 "(msgq_try_lock).");
1103 if (!write_access_) {
1104 throw InterfaceWriteDeniedException(type_,
1106 "Cannot work on message queue on "
1107 "reading instance of an interface (unlock).");
1110 message_queue_->
unlock();
1122 MessageQueue::MessageIterator
1125 if (!write_access_) {
1126 throw InterfaceWriteDeniedException(type_,
1128 "Cannot work on message queue on "
1129 "reading instance of an interface (begin).");
1132 return message_queue_->
begin();
1144 MessageQueue::MessageIterator
1147 if (!write_access_) {
1148 throw InterfaceWriteDeniedException(type_,
1150 "Cannot work on message queue on "
1151 "reading instance of an interface (end).");
1154 return message_queue_->
end();
1166 if (!write_access_) {
1167 throw InterfaceWriteDeniedException(type_,
1169 "Cannot work on message queue on "
1170 "reading instance of an interface (first).");
1172 return message_queue_->
first();
1181 if (!write_access_) {
1182 throw InterfaceWriteDeniedException(type_,
1184 "Cannot work on message queue on "
1185 "reading instance of an interface (pop).");
1188 message_queue_->
pop();
1194 InterfaceFieldIterator
1197 return InterfaceFieldIterator(
this, fieldinfo_list_);
1227 data_mutex_->
lock();
1229 if (buffers_ != NULL) {
1238 throw Exception(errno,
"Resizing buffers for interface %s failed", uid_);
1253 return num_buffers_;
1262 if (buffer >= num_buffers_) {
1267 data_mutex_->
lock();
1269 void *buf = (
char *)buffers_ + buffer *
data_size;
1288 if (buffer >= num_buffers_) {
1289 throw OutOfBoundsException(
"Buffer ID out of bounds", buffer, 0, num_buffers_);
1292 data_mutex_->
lock();
1293 void *buf = (
char *)buffers_ + buffer *
data_size;
1304 if (buffer >= num_buffers_) {
1308 data_mutex_->
lock();
1309 void *buf = (
char *)buffers_ + buffer *
data_size;
1311 *local_read_timestamp_ = *timestamp_;
1326 if (buffer >= num_buffers_) {
1330 data_mutex_->
lock();
1331 void *buf = (
char *)buffers_ + buffer *
data_size;
1345 if (buffer >= num_buffers_) {
1350 void * buf = (
char *)buffers_ + buffer *
data_size;
1363 if (buffer >= num_buffers_) {
1371 void * buf = (
char *)buffers_ + buffer *
data_size;
1372 interface_data_ts_t *buf_ts = (interface_data_ts_t *)buf;
1390 #define xstr(s) str(s)
1391 if ((ec = regcomp(&re,
1392 "^([a-zA-Z0-9]{1," xstr(INTERFACE_TYPE_SIZE_)
"})::"
1393 "([a-zA-Z0-9 _/\\.-]{1," xstr(
1394 INTERFACE_ID_SIZE_)
"})$",
1398 regerror(ec, &re, errbuf, 1024);
1399 throw Exception(
"Failed to created regular expression to parse UID (%s)", errbuf);
1401 regmatch_t matches[3];
1402 if (regexec(&re,
uid, 3, matches, 0) != 0) {
1404 throw Exception(
"Failed to match UID %s, format error.",
uid);
1407 type.assign(&(
uid[matches[1].rm_so]), matches[1].rm_eo - matches[1].rm_so);
1408 id.assign(&(
uid[matches[2].rm_so]), matches[2].rm_eo - matches[2].rm_so);
void lock()
Lock this mutex.
bool empty() const
Check if message queue is empty.
std::list< const char * > get_message_types()
Obtain a list of textual representations of the message types available for this interface.
void unref()
Decrement reference count and conditionally delete this instance.
const void * datachunk() const
Get data chunk.
void set_validity(bool valid)
Mark this interface invalid.
MessageQueue::MessageIterator msgq_end()
Get end iterator for message queue.
const char * type
the type of the message
void msgq_pop()
Erase first message from queue.
void set_time(const timeval *tv)
Sets the time.
bool msgq_empty()
Check if queue is empty.
interface_fieldinfo_t * next
next field, NULL if last
bool try_lock()
Try to lock message queue.
interface_messageinfo_t * next
the next field, NULL if last
InterfaceFieldIterator fields_end()
Invalid iterator.
void flush()
Delete all messages from queue.
int compare_buffers(unsigned int buffer)
Compare buffer to private memory.
const Time * timestamp() const
Get timestamp of last write.
interface_fieldtype_t
Interface field type.
bool is_writer() const
Check if this is a writing instance.
const char * hash_printable() const
Get printable interface hash.
void read()
Read from BlackBoard into local copy.
void set_id(unsigned int message_id)
Set message ID.
Timestamp data, must be present and first entries for each interface data structs!...
const unsigned char * hash() const
Get interface hash.
void read_from_buffer(unsigned int buffer)
Copy data from buffer to private memory.
void lock()
Lock message queue.
virtual bool message_valid(const Message *message) const =0
void pop()
Erase first message from queue.
void unlock()
Unlock message queue.
const char * type() const
Get type of interface.
InterfaceInvalidException(const Interface *interface, const char *method)
Constructor.
void msgq_remove(Message *message)
Remove message from queue.
void add_fieldinfo(interface_fieldtype_t type, const char *name, size_t length, void *value, const char *enumtype=0, const interface_enum_map_t *enum_map=0)
Add an entry to the field info list.
void set_clock(Clock *clock)
Set clock to use for timestamping.
InterfaceInvalidMessageException(const Interface *interface, const Message *message)
Constructor.
int64_t timestamp_sec
time in seconds since Unix epoch
const char * id() const
Get identifier of interface.
void unlock()
Unlock the mutex.
interface_data_ts_t * data_ts
unsigned int mem_serial() const
Get memory serial of interface.
int64_t timestamp_usec
additional time microseconds
void set_timestamp(const Time *t=NULL)
Set timestamp.
void lock_for_read()
Aquire a reader lock.
void mark_data_changed()
Mark data as changed.
unsigned int size() const
Get number of messages in queue.
InterfaceWriteDeniedException(const char *type, const char *id, const char *msg)
Constructor.
MessageIterator end()
Get iterator to element beyond end of message queue list.
std::string writer() const
Get owner name of writing interface instance.
void msgq_lock()
Lock message queue.
const char * enumtype
text representation of enum type
Time buffer_timestamp(unsigned int buffer)
Get time of a buffer.
bool changed() const
Check if data has been changed.
void set_hash(unsigned char *ihash)
Set hash.
std::list< std::string > readers() const
Get owner names of reading interface instances.
Interface field info list.
void append(Message *msg)
Append message to queue.
MessageQueue::MessageIterator msgq_begin()
Get start iterator for message queue.
void remove(const Message *msg)
Remove message from queue.
void lock_for_write()
Aquire a writer lock.
void msgq_append(Message *message)
Enqueue message.
unsigned int msgq_enqueue_copy(Message *message)
Enqueue copy of message at end of queue.
void resize_buffers(unsigned int num_buffers)
Resize buffer array.
bool has_writer() const
Check if there is a writer for the interface.
void set_clock(Clock *clock)
Set clock for this instance.
Message * first()
Get first message from queue.
void msgq_unlock()
Unlock message queue.
unsigned int num_buffers() const
Get number of buffers.
const interface_enum_map_t * enum_map
Map of possible enum values.
static void parse_uid(const char *uid, std::string &type, std::string &id)
Parse UID to type and ID strings.
std::map< int, std::string > interface_enum_map_t
Map of enum integer to string values.
void unlock()
Release the lock.
const char * uid() const
Get unique identifier of interface.
const char * name
Name of this field.
unsigned int num_readers() const
Get the number of readers.
InterfaceFieldIterator fields()
Get iterator over all fields of this interface instance.
const char * owner() const
Get owner of interface.
void set_auto_timestamping(bool enabled)
Enable or disable automated timestamping.
size_t length
Length of field (array, string)
unsigned short serial() const
Get instance serial of interface.
unsigned int msgq_size()
Get size of message queue.
unsigned int num_fields()
Get the number of fields in the interface.
virtual Message * clone() const
Clone this message.
unsigned int id() const
Get message ID.
static Clock * instance()
Clock initializer.
Time & stamp()
Set this time to the current time.
virtual ~Interface()
Destructor.
void copy_private_to_buffer(unsigned int buffer)
Copy data from private memory to buffer.
void msgq_flush()
Flush all messages.
unsigned int datasize() const
Get data size.
unsigned int msgq_enqueue(Message *message)
Enqueue message at end of queue.
bool oftype(const char *interface_type) const
Check if interface is of given type.
void set_from_chunk(void *chunk)
Set from a raw data chunk.
void get_timestamp(long &sec, long &usec) const
InterfaceMessageEnqueueException(const char *type, const char *id)
Constructor.
Message * msgq_first()
Get the first message from the message queue.
void add_messageinfo(const char *name)
Add an entry to the message info list.
void write()
Write from local copy into BlackBoard memory.
interface_fieldtype_t type
type of this field
size_t hash_size() const
Get size of interface hash.
bool is_valid() const
Check validity of interface.
void copy_shared_to_buffer(unsigned int buffer)
Copy data from private memory to buffer.
MessageIterator begin()
Get iterator to first element in message queue.
bool msgq_try_lock()
Try to lock message queue.
bool operator==(Interface &comp) const
Check equality of two interfaces.
void * value
Current value of this field.