111 const std::string& message) :
117 memcpy(this->
buffer, static_cast<const JPrefix_t*>(
this),
sizeof(
JPrefix_t));
119 memcpy(this->
buffer +
sizeof(
JPrefix_t), message.data(), message.size());
130 static_cast<JPrefix_t&
>(*this) = message;
355 for (istringstream
is(subscription);
is >> c >> tag; ) {
415 queue.push_back(message);
429 for (std::deque<JDispatch>::iterator i =
queue.begin(); i !=
queue.end(); ) {
473 for (
iterator i = this->begin(); i != this->end(); ++i) {
484 for (
iterator i = this->begin(); i != this->end(); ++i) {
500 return out <<
"(" << message.
getTag() <<
"," << message.
size() <<
")";
556 int main(
int argc,
char* argv[])
581 catch(
const exception &error) {
582 FATAL(error.what() << endl);
586 JServerSocket server(port, backlog);
588 JClientList clientList;
591 DEBUG(
"Port " << setw(10) << port << endl);
599 select.setReaderMask(server);
601 for (JClientList::iterator
client = clientList.begin();
client != clientList.end(); ++
client) {
603 if (!
client->in.isReady()) {
604 select.setReaderMask(*
client);
607 if (
client->out.isReset()) {
609 if (!
client->queue.empty()) {
614 client->decrementRequest();
616 select.setWriterMask(*
client);
619 }
else if (
client->out.isBusy()) {
621 select.setWriterMask(*
client);
625 if (select(timeout_us) > 0) {
627 for (JClientList::iterator
client = clientList.begin();
client != clientList.end(); ) {
631 if (select.hasReaderMask(*
client)) {
636 catch(
const exception& error) {
638 ERROR(
"Remove (3) client" << *
client <<
"<" <<
client->getNickname() <<
">: " << error.what() << endl);
640 if (
client->getNickname() !=
"") {
651 DEBUG(
"Client" << *
client <<
".read" << static_cast<const JSocketInputBuffer&>(
client->in) << endl);
654 if (
client->in.isReady()) {
666 client->setSubscription(
string(
client->in.getRemainingData(),
client->in.getRemainingSize()));
670 client->setNickname(
string(
client->in.getRemainingData(),
client->in.getRemainingSize()));
676 client->incrementRequest();
684 string nick_name(
client->in.getRemainingData(),
client->in.getRemainingSize());
687 for (JClientList::iterator i = clientList.begin(); i != clientList.end(); ++i) {
688 if (i->getNickname() == nick_name) {
689 buffer +=
" " + i->getHostname();
693 JControlHost socket(*
client);
695 socket.PutFullString(
DISPTAG_WhereIs, buffer.substr(buffer.empty() ? 0 : 1));
707 istringstream
is(
string(
client->in.getRemainingData(),
client->in.getRemainingSize()));
719 clientList.add(JDispatch(
client->in.prefix,
client->in.data()));
732 if (select.hasWriterMask(*
client)) {
736 DEBUG(
"Client" << *
client <<
".write" << static_cast<const JSocketStatus&>(
client->out) << endl);
738 if (
client->out.isReady()) {
740 client->queue.pop_front();
746 catch(
const exception& error) {
748 DEBUG(
"Remove (2) client" << *
client <<
"<" <<
client->getNickname() <<
">: " << error.what() << endl);
750 if (
client->getNickname() !=
"") {
760 if (select.hasReaderMask(server)) {
764 socket.accept(server.getFileDescriptor());
769 socket.setKeepAlive (
true);
770 socket.setNonBlocking(
true);
772 DEBUG(
"New client" << socket << endl);
774 clientList.push_back(JClient(socket));
JSocketNonblockingWriter out
writer for outgoing messages
bool setSubscription(const std::string &subscription)
Set subcription.
Utility class to parse command line options.
static const JTag DISPTAG_Subscribe("_Subscri")
Special ControlHost tags.
std::set< JTag > subscriptionAll
int getSizeOfPacket(const KM3NETDAQ::JDAQAbstractPreamble &preamble)
Get size of packeet.
static const int DISPATCH_PORT
Default ControlHost port.
JDispatch & operator=(const JDispatch &message)
Assignment operator.
static const JTag DISPTAG_Died("Died")
Data structure of a ControlHost message.
void setSize(const long long int length)
Set size.
JSocketBuffer< const char > JSocketOutputBuffer
void setRequestAll()
Set no request.
ControlHost client manager.
std::deque< JDispatch > queue
queue for outgoing messages
void release()
Release memory.
Auxiliary class for non-blocking socket I/O.
List of ControlHost client managers.
std::set< JTag > subscriptionAny
static const JTag DISPTAG_MyId("_MyId")
void attach(const JSharedCounter &object)
Attach this counter to given shared counter object.
void setNickname(const std::string &nick_name)
Set nick name.
static const JTag DISPTAG_WhereIs("_WhereIs")
bool checkRequest() const
Check request.
bool checkSubscription(const JPrefix_t &prefix) const
Check subscription for given prefix.
JClient()
Default constructor.
const std::string & getNickname() const
Get nick name.
void decrementRequest()
Decrement request by one.
bool checkSubscriptionAll(const JPrefix_t &prefix) const
Check subscription for given prefix.
static long long int MEMORY_LIMIT
Limit size of data [Bytes].
int getSize() const
Get size.
JClientList()
Default constructor.
then echo Test string reversed by client(hit< return > to continue)." JProcess -c "JEcho-r" -C fi if (( 1 ))
static void release(T *p)
Release memory.
bool checkSubscriptionAny(const JPrefix_t &prefix) const
Check subscription for given prefix.
int getFileDescriptor() const
Get file descriptor.
JSocketInputChannel_t in
reader for incoming messages
int getCounter() const
Get number of I/O attempts.
std::ostream & operator<<(std::ostream &out, const JDispatch &message)
Print message.
Non-blocking socket writer.
void incrementRequest()
Increment request by one.
#define make_field(A,...)
macro to convert parameter to JParserTemplateElement object
bool isReady() const
Check ready status.
void add(const JDispatch &message)
Add message to client queues depending on subscription of each client.
Memory management class for create/release policy based on malloc()/free().
Exception for failure of memory allocation.
static const size_t buffer_size
JDispatch()
Default constructor.
void drop()
Drop all messages for which the client has not the 'all' subscription.
JDispatch(const JPrefix_t &prefix, const char *data)
Constructor.
JDispatch(const JDispatch &message)
Copy constructor.
void create()
Allocate memory.
void drop()
Drop all messages from client queues for which the client has not the 'all' subscription.
Exception for ControlHost.
JClient(const JSocket &socket)
Constructor.
const JTag & getTag() const
Get tag.
static bool maybe_special(const JTag &tag)
Check special ControlHost tags.
General purpose messaging.
void initialise()
Initialise counter.
static const JTag DISPTAG_Born("Born")
static long long int MEMORY_TOTAL
Total size of data [Bytes].
JMalloc< char > JMemory_t
static unsigned int QUEUE_LIMIT
Maximum number of messages in queue.
static const long long int GIGABYTE
Number of bytes in a mega-byte.
const char * data() const
Get data.
Utility class to parse command line options.
void add(const JDispatch &message)
Add message to client queues depending on subscription of each client.
unsigned long long int getRAM()
Get RAM of this CPU.
static const JTag DISPTAG_Gime("_Gime")
int getSize() const
Get size of pending data.
JSocketInputChannel< JPrefix_t > JSocketInputChannel_t
void setSizeOfPacket(const int size, JPrefix_t &prefix)
Set total size of internet packet.
static const JTag DISPTAG_Debug("_Debug")
static const JTag DISPTAG_Always("_Always")
Base class for memory management.
Auxiliary class for non-blocking socket I/O.
static T * create()
Create object in memory.
bool isBusy() const
Check busy status.
int size() const
Get size.
JDispatch(const JTag &tag, const std::string &message)
Constructor.
#define DEBUG(A)
Message macros.
int main(int argc, char *argv[])