@@ -125,13 +125,11 @@ static void reconnect(Shub* shub)
125125 }while (rc < 0 && errno == EINTR );
126126
127127if (rc >=0 || errno == EINPROGRESS ) {
128- if (rc >=0 ) {
129- }
130128break ;
131129 }
132130 }
133131if (rc < 0 ) {
134- if (errno != ENOENT && errno != ECONNREFUSED ) {
132+ if (errno != ENOENT && errno != ECONNREFUSED && errno != EINPROGRESS ) {
135133shub -> params -> error_handler ("Connection can not be establish" ,SHUB_FATAL_ERROR );
136134 }
137135if (max_attempts -- != 0 ) {
@@ -187,6 +185,7 @@ void ShubInitialize(Shub* shub, ShubParams* params)
187185FD_ZERO (& shub -> inset );
188186FD_SET (shub -> input ,& shub -> inset );
189187
188+ shub -> output = -1 ;
190189reconnect (shub );
191190
192191shub -> in_buffer = malloc (params -> buffer_size );
@@ -207,52 +206,58 @@ void ShubLoop(Shub* shub)
207206while (1 ) {
208207fd_set events ;
209208struct timeval tm ;
210- int i ,max_fd , rc ;
211- unsigned int n , size ;
209+ int i ,rc ;
210+ int max_fd = shub -> max_fd ;
212211
213212tm .tv_sec = shub -> params -> delay /1000 ;
214213tm .tv_usec = shub -> params -> delay %1000 * 1000 ;
215214
216215events = shub -> inset ;
217- rc = select (shub -> max_fd + 1 ,& events ,NULL ,NULL ,shub -> in_buffer_used == 0 ?NULL :& tm );
216+ rc = select (max_fd + 1 ,& events ,NULL ,NULL ,shub -> in_buffer_used == 0 ?NULL :& tm );
218217if (rc < 0 ) {
219218if (errno != EINTR ) {
220219shub -> params -> error_handler ("Select failed" ,SHUB_RECOVERABLE_ERROR );
221220recovery (shub );
222221 }
223222 }else {
224223if (rc > 0 ) {
225- for (i = 0 , max_fd = shub -> max_fd ;i <=max_fd ;i ++ ) {
224+ for (i = 0 ;i <=max_fd ;i ++ ) {
226225if (FD_ISSET (i ,& events )) {
227- if (i == shub -> input ) {
226+ if (i == shub -> input ) {/* accept incomming connection */
228227int s = accept (i ,NULL ,NULL );
229228if (s < 0 ) {
230229shub -> params -> error_handler ("Failed to accept socket" ,SHUB_RECOVERABLE_ERROR );
231230 }else {
232- if (s > max_fd ) {
231+ if (s > shub -> max_fd ) {
233232shub -> max_fd = s ;
234233 }
235234FD_SET (s ,& shub -> inset );
236235 }
237- }else if (i == shub -> output ) {
236+ }else if (i == shub -> output ) {/* receive response from server */
237+ /* try to read as much as possible */
238238int available = recv (shub -> output ,shub -> out_buffer + shub -> out_buffer_used ,buffer_size - shub -> out_buffer_used ,0 );
239239int pos = 0 ;
240240if (available <=0 ) {
241241shub -> params -> error_handler ("Failed to read inet socket" ,SHUB_RECOVERABLE_ERROR );
242242reconnect (shub );
243+ continue ;
243244 }
244245shub -> out_buffer_used += available ;
246+
247+ /* loop through all received responses */
245248while (pos + sizeof (ShubMessageHdr ) <=shub -> out_buffer_used ) {
246- ShubMessageHdr * hdr = (ShubMessageHdr * )( shub -> out_buffer + pos ) ;
249+ ShubMessageHdr * hdr = (ShubMessageHdr * )& shub -> out_buffer [ pos ] ;
247250int chan = hdr -> chan ;
248- n = pos + sizeof (ShubMessageHdr )+ hdr -> size <=shub -> out_buffer_used ?hdr -> size + sizeof (ShubMessageHdr ) :shub -> out_buffer_used - pos ;
251+ unsignedint n = pos + sizeof (ShubMessageHdr )+ hdr -> size <=shub -> out_buffer_used
252+ ?hdr -> size + sizeof (ShubMessageHdr )
253+ :shub -> out_buffer_used - pos ;
249254if (!write_socket (chan , (char * )hdr ,n )) {
250255shub -> params -> error_handler ("Failed to write to local socket" ,SHUB_RECOVERABLE_ERROR );
251256close_socket (shub ,chan );
252257chan = -1 ;
253258 }
254- /* read rest of message if it doesn't fit in buffer */
255259if (n != hdr -> size + sizeof (ShubMessageHdr )) {
260+ /* read rest of message if it doesn't fit in the buffer */
256261int tail = hdr -> size + sizeof (ShubMessageHdr )- n ;
257262do {
258263n = tail < buffer_size ?tail :buffer_size ;
@@ -274,56 +279,73 @@ void ShubLoop(Shub* shub)
274279 }
275280pos += n ;
276281 }
282+ /* Move partly fetched message header (if any) to the beginning of byffer */
277283memcpy (shub -> out_buffer ,shub -> out_buffer + pos ,shub -> out_buffer_used - pos );
278284shub -> out_buffer_used -= pos ;
279- }else {
285+ }else {/* receive request from client */
280286ShubMessageHdr * hdr = (ShubMessageHdr * )& shub -> in_buffer [shub -> in_buffer_used ];
281- if (!read_socket (i , (char * )hdr ,sizeof (ShubMessageHdr ))) {
287+ int chan = i ;
288+ if (!read_socket (chan , (char * )hdr ,sizeof (ShubMessageHdr ))) {/* fetch message header */
282289shub -> params -> error_handler ("Failed to read local socket" ,SHUB_RECOVERABLE_ERROR );
283290close_socket (shub ,i );
284291 }else {
285- size = hdr -> size ;
286- hdr -> chan = i ;
292+ unsigned int size = hdr -> size ;
293+ hdr -> chan = chan ; /* remember socket descriptor from which this message was read */
287294if (size + shub -> in_buffer_used + sizeof (ShubMessageHdr )> buffer_size ) {
288- if (shub -> in_buffer_used != 0 ) {
295+ /* message doesn't completely fit in buffer */
296+ if (shub -> in_buffer_used != 0 ) {/* if buffer is not empty...*/
297+ /* ... then send it */
289298while (!write_socket (shub -> output ,shub -> in_buffer ,shub -> in_buffer_used )) {
290299shub -> params -> error_handler ("Failed to write to inet socket" ,SHUB_RECOVERABLE_ERROR );
291300reconnect (shub );
292301 }
302+ /* move received message header to the beginning of the buffer */
293303memcpy (shub -> in_buffer ,shub -> in_buffer + shub -> in_buffer_used ,sizeof (ShubMessageHdr ));
294304shub -> in_buffer_used = 0 ;
295305 }
296306 }
297307shub -> in_buffer_used += sizeof (ShubMessageHdr );
298308
299- while ( 1 ) {
309+ do {
300310unsignedint n = size + shub -> in_buffer_used > buffer_size ?buffer_size - shub -> in_buffer_used :size ;
301- if (!read_socket (i ,shub -> in_buffer + shub -> in_buffer_used ,n )) {
311+ /* fetch message body */
312+ if (chan >=0 && !read_socket (chan ,shub -> in_buffer + shub -> in_buffer_used ,n )) {
302313shub -> params -> error_handler ("Failed to read local socket" ,SHUB_RECOVERABLE_ERROR );
303- close_socket (shub ,i );
304- break ;
305- }else {
306- if (n != size ) {
307- while (!write_socket (shub -> output ,shub -> in_buffer ,n )) {
308- shub -> params -> error_handler ("Failed to write to inet socket" ,SHUB_RECOVERABLE_ERROR );
309- reconnect (shub );
310- }
311- size -= n ;
312- shub -> in_buffer_used = 0 ;
313- }else {
314- shub -> in_buffer_used += n ;
314+ close_socket (shub ,chan );
315+ if (hdr != NULL ) {/* if message header is not yet sent to the server... */
316+ /* ... then skip this message */
317+ shub -> in_buffer_used = (char * )hdr - shub -> in_buffer ;
315318break ;
319+ }else {/* if message was partly sent to the server, we can not skip it, so we have to send garbage to the server */
320+ chan = -1 ;/* do not try to read rest of body of this message */
316321 }
322+ }
323+ shub -> in_buffer_used += n ;
324+ size -= n ;
325+ /* if there is no more free space in the buffer to receive new message header... */
326+ if (shub -> in_buffer_used + sizeof (ShubMessageHdr )> buffer_size ) {
327+
328+ /* ... then send buffer to the server */
329+ while (!write_socket (shub -> output ,shub -> in_buffer ,shub -> in_buffer_used )) {
330+ shub -> params -> error_handler ("Failed to write to inet socket" ,SHUB_RECOVERABLE_ERROR );
331+ reconnect (shub );
332+ }
333+ hdr = NULL ;/* message is partly sent to the server: can not skip it any more */
334+ shub -> in_buffer_used = 0 ;
317335 }
318- }
336+ }while ( size != 0 ); /* repeat until all message body is received */
319337 }
320338 }
321339 }
322340 }
323- }else if (shub -> in_buffer_used != 0 ) {
324- while (!write_socket (shub -> output ,shub -> in_buffer ,shub -> in_buffer_used )) {
325- shub -> params -> error_handler ("Failed to write to inet socket" ,SHUB_RECOVERABLE_ERROR );
326- reconnect (shub );
341+ }else {/* timeout expired */
342+ if (shub -> in_buffer_used != 0 ) {/* if buffer is not empty... */
343+ /* ...then send it */
344+ while (!write_socket (shub -> output ,shub -> in_buffer ,shub -> in_buffer_used )) {
345+ shub -> params -> error_handler ("Failed to write to inet socket" ,SHUB_RECOVERABLE_ERROR );
346+ reconnect (shub );
347+ }
348+ shub -> in_buffer_used = 0 ;
327349 }
328350 }
329351 }