taskrambler / libtrcomm · Commits

Commit 18fcc5872344dd080b4f12cd32833cb634907e32

Authored by Georg Hopp
1 parent ec07940a

adapt on separation of issue and emit of events

Inline Side-by-side
Showing 6 changed files with 60 additions and 44 deletions
@@ -87,30 +87,33 @@ TR_commManagerPollSelect(void * _this, TR_Event event, int timeout) @@ -87,30 +87,33 @@ TR_commManagerPollSelect(void * _this, TR_Event event, int timeout)
87 TR_CommEndPoint endpoint = cmgr->endpoints[i]; 87 TR_CommEndPoint endpoint = cmgr->endpoints[i];
88 88
89 if ((this->fds[i].revents & POLLIN) == POLLIN) { 89 if ((this->fds[i].revents & POLLIN) == POLLIN) {
  90 + TR_Event event;
  91 +
90 nevents--; 92 nevents--;
91 if (TR_INSTANCE_OF(TR_TcpSocket, endpoint->transport) 93 if (TR_INSTANCE_OF(TR_TcpSocket, endpoint->transport)
92 && ((TR_TcpSocket)endpoint->transport)->listen) { 94 && ((TR_TcpSocket)endpoint->transport)->listen) {
93 - TR_eventHandlerIssueEvent(  
94 - (TR_EventHandler)this, 95 + event = TR_eventSubjectEmit(
95 (TR_EventSubject)endpoint, 96 (TR_EventSubject)endpoint,
96 TR_CET_EVENT_ACC_READY, 97 TR_CET_EVENT_ACC_READY,
97 NULL); 98 NULL);
98 } else { 99 } else {
99 - TR_eventHandlerIssueEvent(  
100 - (TR_EventHandler)this, 100 + event = TR_eventSubjectEmit(
101 (TR_EventSubject)endpoint, 101 (TR_EventSubject)endpoint,
102 TR_CEP_EVENT_READ_READY, 102 TR_CEP_EVENT_READ_READY,
103 NULL); 103 NULL);
104 } 104 }
  105 +
  106 + TR_eventHandlerIssueEvent((TR_EventHandler)this, event);
105 } 107 }
106 108
107 if ((this->fds[i].revents & POLLOUT) == POLLOUT) { 109 if ((this->fds[i].revents & POLLOUT) == POLLOUT) {
108 nevents--; 110 nevents--;
109 TR_eventHandlerIssueEvent( 111 TR_eventHandlerIssueEvent(
110 (TR_EventHandler)this, 112 (TR_EventHandler)this,
111 - (TR_EventSubject)endpoint,  
112 - TR_CEP_EVENT_WRITE_READY,  
113 - NULL); 113 + TR_eventSubjectEmit(
  114 + (TR_EventSubject)endpoint,
  115 + TR_CEP_EVENT_WRITE_READY,
  116 + NULL));
114 } 117 }
115 118
116 if (nevents <= 0) break; 119 if (nevents <= 0) break;
@@ -38,9 +38,10 @@ TR_commManagerShutdown(void * _this, TR_Event event) @@ -38,9 +38,10 @@ TR_commManagerShutdown(void * _this, TR_Event event)
38 if (this->endpoints[i]) { 38 if (this->endpoints[i]) {
39 TR_eventHandlerIssueEvent( 39 TR_eventHandlerIssueEvent(
40 (TR_EventHandler)_this, 40 (TR_EventHandler)_this,
41 - (TR_EventSubject)this->endpoints[i],  
42 - TR_CEP_EVENT_CLOSE,  
43 - NULL); 41 + TR_eventSubjectEmit(
  42 + (TR_EventSubject)this->endpoints[i],
  43 + TR_CEP_EVENT_CLOSE,
  44 + NULL));
44 } 45 }
45 } 46 }
46 47
@@ -67,9 +67,10 @@ connectorAccept(void * _this, TR_Event event) @@ -67,9 +67,10 @@ connectorAccept(void * _this, TR_Event event)
67 8192); 67 8192);
68 TR_eventHandlerIssueEvent( 68 TR_eventHandlerIssueEvent(
69 (TR_EventHandler)this, 69 (TR_EventHandler)this,
70 - (TR_EventSubject)new_con,  
71 - TR_CON_EVENT_NEW_CON,  
72 - NULL); 70 + TR_eventSubjectEmit(
  71 + (TR_EventSubject)new_con,
  72 + TR_CON_EVENT_NEW_CON,
  73 + NULL));
73 socket = TR_socketAccept((TR_TcpSocket)connection->transport); 74 socket = TR_socketAccept((TR_TcpSocket)connection->transport);
74 } 75 }
75 76
@@ -83,9 +83,10 @@ TR_commManagerDisableWrite(void * _this, TR_Event event) @@ -83,9 +83,10 @@ TR_commManagerDisableWrite(void * _this, TR_Event event)
83 if (TR_socketFinRd(endpoint->transport)) { 83 if (TR_socketFinRd(endpoint->transport)) {
84 TR_eventHandlerIssueEvent( 84 TR_eventHandlerIssueEvent(
85 this, 85 this,
86 - event->subject,  
87 - TR_CEP_EVENT_SHUT_READ,  
88 - NULL); 86 + TR_eventSubjectEmit(
  87 + event->subject,
  88 + TR_CEP_EVENT_SHUT_READ,
  89 + NULL));
89 } 90 }
90 91
91 return 1; 92 return 1;
@@ -113,16 +114,18 @@ TR_commManagerShutdownRead(void * _this, TR_Event event) @@ -113,16 +114,18 @@ TR_commManagerShutdownRead(void * _this, TR_Event event)
113 // close 114 // close
114 TR_eventHandlerIssueEvent( 115 TR_eventHandlerIssueEvent(
115 (TR_EventHandler)_this, 116 (TR_EventHandler)_this,
116 - event->subject,  
117 - TR_CEP_EVENT_CLOSE,  
118 - NULL); 117 + TR_eventSubjectEmit(
  118 + event->subject,
  119 + TR_CEP_EVENT_CLOSE,
  120 + NULL));
119 } else if (TR_cepHasPendingData((TR_CommEndPoint)event->subject)) { 121 } else if (TR_cepHasPendingData((TR_CommEndPoint)event->subject)) {
120 // handle pending data... close is issued from disableWrite 122 // handle pending data... close is issued from disableWrite
121 TR_eventHandlerIssueEvent( 123 TR_eventHandlerIssueEvent(
122 (TR_EventHandler)_this, 124 (TR_EventHandler)_this,
123 - event->subject,  
124 - TR_CEP_EVENT_CLOSE,  
125 - NULL); 125 + TR_eventSubjectEmit(
  126 + event->subject,
  127 + TR_CEP_EVENT_CLOSE,
  128 + NULL));
126 } else { 129 } else {
127 TR_cepSetClose((TR_CommEndPoint)event->subject); 130 TR_cepSetClose((TR_CommEndPoint)event->subject);
128 } 131 }
@@ -138,9 +141,10 @@ TR_commManagerShutdownWrite(void * _this, TR_Event event) @@ -138,9 +141,10 @@ TR_commManagerShutdownWrite(void * _this, TR_Event event)
138 if (TR_socketFinRd(((TR_CommEndPoint)event->subject)->transport)) { 141 if (TR_socketFinRd(((TR_CommEndPoint)event->subject)->transport)) {
139 TR_eventHandlerIssueEvent( 142 TR_eventHandlerIssueEvent(
140 (TR_EventHandler)_this, 143 (TR_EventHandler)_this,
141 - event->subject,  
142 - TR_CEP_EVENT_CLOSE,  
143 - NULL); 144 + TR_eventSubjectEmit(
  145 + event->subject,
  146 + TR_CEP_EVENT_CLOSE,
  147 + NULL));
144 } 148 }
145 149
146 return 0; 150 return 0;
@@ -54,25 +54,28 @@ ioHandlerRead(void * _this, TR_Event event) @@ -54,25 +54,28 @@ ioHandlerRead(void * _this, TR_Event event)
54 case -1: 54 case -1:
55 TR_eventHandlerIssueEvent( 55 TR_eventHandlerIssueEvent(
56 (TR_EventHandler)_this, 56 (TR_EventHandler)_this,
57 - event->subject,  
58 - TR_CEP_EVENT_CLOSE,  
59 - NULL); 57 + TR_eventSubjectEmit(
  58 + event->subject,
  59 + TR_CEP_EVENT_CLOSE,
  60 + NULL));
60 break; 61 break;
61 62
62 case -2: 63 case -2:
63 TR_eventHandlerIssueEvent( 64 TR_eventHandlerIssueEvent(
64 (TR_EventHandler)_this, 65 (TR_EventHandler)_this,
65 - event->subject,  
66 - TR_CEP_EVENT_SHUT_READ,  
67 - NULL); 66 + TR_eventSubjectEmit(
  67 + event->subject,
  68 + TR_CEP_EVENT_SHUT_READ,
  69 + NULL));
68 break; 70 break;
69 71
70 case TRUE: 72 case TRUE:
71 TR_eventHandlerIssueEvent( 73 TR_eventHandlerIssueEvent(
72 (TR_EventHandler)_this, 74 (TR_EventHandler)_this,
73 - event->subject,  
74 - TR_CEP_EVENT_NEW_DATA,  
75 - NULL); 75 + TR_eventSubjectEmit(
  76 + event->subject,
  77 + TR_CEP_EVENT_NEW_DATA,
  78 + NULL));
76 break; 79 break;
77 } 80 }
78 81
@@ -86,19 +89,21 @@ ioHandlerWrite(void * _this, TR_Event event) @@ -86,19 +89,21 @@ ioHandlerWrite(void * _this, TR_Event event)
86 TR_CommEndPoint endpoint = (TR_CommEndPoint)event->subject; 89 TR_CommEndPoint endpoint = (TR_CommEndPoint)event->subject;
87 90
88 if (TR_cepWriteBuffered(endpoint)) { 91 if (TR_cepWriteBuffered(endpoint)) {
  92 + TR_Event event;
  93 +
89 if (TR_cepHasPendingData(endpoint)) { 94 if (TR_cepHasPendingData(endpoint)) {
90 - TR_eventHandlerIssueEvent(  
91 - (TR_EventHandler)_this, 95 + event = TR_eventSubjectEmit(
92 event->subject, 96 event->subject,
93 TR_CEP_EVENT_PENDING_DATA, 97 TR_CEP_EVENT_PENDING_DATA,
94 NULL); 98 NULL);
95 } else { 99 } else {
96 - TR_eventHandlerIssueEvent(  
97 - (TR_EventHandler)_this, 100 + event = TR_eventSubjectEmit(
98 event->subject, 101 event->subject,
99 TR_CEP_EVENT_END_DATA, 102 TR_CEP_EVENT_END_DATA,
100 NULL); 103 NULL);
101 } 104 }
  105 +
  106 + TR_eventHandlerIssueEvent((TR_EventHandler)_this, event);
102 } 107 }
103 108
104 return TRUE; 109 return TRUE;
@@ -56,9 +56,10 @@ protocolHandlerParse(void * _this, TR_Event event) @@ -56,9 +56,10 @@ protocolHandlerParse(void * _this, TR_Event event)
56 if (message) { 56 if (message) {
57 TR_eventHandlerIssueEvent( 57 TR_eventHandlerIssueEvent(
58 (TR_EventHandler)_this, 58 (TR_EventHandler)_this,
59 - event->subject,  
60 - TR_CEP_EVENT_NEW_MSG,  
61 - message); 59 + TR_eventSubjectEmit(
  60 + event->subject,
  61 + TR_CEP_EVENT_NEW_MSG,
  62 + message));
62 63
63 if (message->close) { 64 if (message->close) {
64 // also check that we are a response. Well this is how it is done 65 // also check that we are a response. Well this is how it is done
@@ -86,9 +87,10 @@ protocolHandlerCompose(void * _this, TR_Event event) @@ -86,9 +87,10 @@ protocolHandlerCompose(void * _this, TR_Event event)
86 if (TR_cepCompose(endpoint, message)) { 87 if (TR_cepCompose(endpoint, message)) {
87 TR_eventHandlerIssueEvent( 88 TR_eventHandlerIssueEvent(
88 (TR_EventHandler)_this, 89 (TR_EventHandler)_this,
89 - event->subject,  
90 - TR_CEP_EVENT_WRITE_READY,  
91 - NULL); 90 + TR_eventSubjectEmit(
  91 + event->subject,
  92 + TR_CEP_EVENT_WRITE_READY,
  93 + NULL));
92 TR_delete(message); 94 TR_delete(message);
93 } 95 }
94 96
Please register or login to post a comment