~ubuntu-branches/ubuntu/raring/workrave/raring
» Revision
12
: .pc/04_disconnect_bugs.patch/backend/src/DistributionSocketLink.cc
« back to all changes in this revision
Viewing changes to .pc/04_disconnect_bugs.patch/backend/src/DistributionSocketLink.cc
- browse files at revision 12
- compare with another revision
- download diff
- download tarball
- view history from revision 12
- Committer: Bazaar Package Importer
- Author(s): Francois Marier
- Date: 2011-03-25 17:17:18 UTC
- mfrom: (1.2.8 upstream)
- Revision ID: james.westby@ubuntu.com-20110325171718-3znn7qnfgmx20iyw
Tags: 1.9.4-1
* New upstream release
- drop all Debian patches (applied upstream)
* Update location of .xpm menu icon
* Remove empty /usr/include directory after building
* Remove obsolete breaks/replaces from debian/control
- drop all Debian patches (applied upstream)
* Update location of .xpm menu icon
* Remove empty /usr/include directory after building
* Remove obsolete breaks/replaces from debian/control
- files added:
- frontend/common/share/images/workrave-sheep.svg
- frontend/gtkmm/src/gnome_applet3
- frontend/gtkmm/win32/setup/src
- files removed:
- .pc
- .pc/01_typo.patch
- .pc/01_typo.patch/frontend
- .pc/01_typo.patch/frontend/gtkmm
- .pc/01_typo.patch/frontend/gtkmm/src
- .pc/02_network_connection_fix.patch
- .pc/02_network_connection_fix.patch/backend
- .pc/02_network_connection_fix.patch/backend/src
- .pc/03_disconnect_crash.patch
- .pc/03_disconnect_crash.patch/backend
- .pc/03_disconnect_crash.patch/backend/src
- .pc/04_disconnect_bugs.patch
- .pc/04_disconnect_bugs.patch/backend
- .pc/04_disconnect_bugs.patch/backend/src
- debian/patches
- frontend/common/src/gnome
- files modified:
- Makefile.in
added
removed
.pc/04_disconnect_bugs.patch/backend/src/DistributionSocketLink.cc
1
// DistributionSocketLink.cc
2
//
3
// Copyright (C) 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2010, 2011 Rob Caelers <robc@krandor.org>
4
// All rights reserved.
5
//
6
// This program is free software: you can redistribute it and/or modify
7
// it under the terms of the GNU General Public License as published by
8
// the Free Software Foundation, either version 3 of the License, or
9
// (at your option) any later version.
10
//
11
// This program is distributed in the hope that it will be useful,
12
// but WITHOUT ANY WARRANTY; without even the implied warranty of
13
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14
// GNU General Public License for more details.
15
//
16
// You should have received a copy of the GNU General Public License
17
// along with this program. If not, see <http://www.gnu.org/licenses/>.
18
19
20
#ifdef HAVE_CONFIG_H
21
#include "config.h"
22
#endif
23
24
#ifdef HAVE_DISTRIBUTION
25
26
#include "debug.hh"
27
#include <assert.h>
28
29
#include <iostream>
30
#include <fstream>
31
32
#include "nls.h"
33
34
#include <stdio.h>
35
#include <stdlib.h>
36
#include <string.h>
37
38
#include <signal.h>
39
40
#include "Configurator.hh"
41
#include "CoreConfig.hh"
42
43
#include "DistributionManager.hh"
44
#include "DistributionLink.hh"
45
#include "DistributionSocketLink.hh"
46
47
#include "Util.hh"
48
49
using namespace std;
50
51
//! Construct a new socket link.
52
/*!
53
* \param conf Configurator to use.
54
*/
55
DistributionSocketLink::DistributionSocketLink(Configurator *conf) :
56
dist_manager(NULL),
57
configurator(conf),
58
username(NULL),
59
password(NULL),
60
master_client(NULL),
61
i_am_master(false),
62
master_locked(false),
63
server_port(DEFAULT_PORT),
64
server_socket(NULL),
65
network_enabled(false),
66
server_enabled(false),
67
reconnect_attempts(DEFAULT_ATTEMPTS),
68
reconnect_interval(DEFAULT_INTERVAL),
69
heartbeat_count(0)
70
{
71
socket_driver = SocketDriver::create();
72
init_my_id();
73
}
74
75
76
//! Destructs the socket link.
77
DistributionSocketLink::~DistributionSocketLink()
78
{
79
remove_client(NULL);
80
81
g_free(username);
82
g_free(password);
83
delete server_socket;
84
delete socket_driver;
85
}
86
87
88
//! Initialize the link.
89
void
90
DistributionSocketLink::init()
91
{
92
TRACE_ENTER("DistributionSocketLink::init");
93
94
// I'm master.
95
master_client = NULL;
96
i_am_master = true;
97
98
// Read all tcp link configuration.
99
read_configuration();
100
configurator->add_listener(CoreConfig::CFG_KEY_DISTRIBUTION_TCP, this);
101
102
TRACE_EXIT();
103
}
104
105
106
//! Periodic heartbeat.
107
void
108
DistributionSocketLink::heartbeat()
109
{
110
if (server_enabled)
111
{
112
heartbeat_count++;
113
114
time_t current_time = time(NULL);
115
116
// See if we have some clients that need reconncting.
117
list<Client *>::iterator i = clients.begin();
118
while (i != clients.end())
119
{
120
Client *c = *i;
121
if (c->type == CLIENTTYPE_DIRECT &&
122
c->reconnect_count > 0 &&
123
c->reconnect_time != 0 && current_time >= c->reconnect_time &&
124
c->hostname != NULL)
125
{
126
c->reconnect_count--;
127
c->reconnect_time = 0;
128
129
dist_manager->log(_("Reconnecting to %s."),
130
c->id == NULL ? "Unknown" : c->id);
131
132
if (c->socket != NULL)
133
{
134
c->socket->close();
135
delete c->socket;
136
}
137
138
ISocket *socket = socket_driver->create_socket();
139
socket->set_data(c);
140
socket->set_listener(this);
141
socket->connect(c->hostname, c->port);
142
143
c->socket = socket;
144
}
145
i++;
146
}
147
148
// Periodically distribute state, in case the master crashes.
149
if (heartbeat_count % 30 == 0 && i_am_master)
150
{
151
send_client_message(DCMT_MASTER);
152
}
153
}
154
}
155
156
157
//! Initializes the network wrapper.
158
void
159
DistributionSocketLink::init_my_id()
160
{
161
TRACE_ENTER("DistributionSocketLink::init_my_id");
162
bool ok = false;
163
string idfilename = Util::get_home_directory() + "id";
164
165
if (Util::file_exists(idfilename))
166
{
167
ifstream file(idfilename.c_str());
168
169
if (file)
170
{
171
string id_str;
172
file >> id_str;
173
174
if (id_str.length() == WRID::STR_LENGTH)
175
{
176
ok = my_id.set(id_str);
177
}
178
file.close();
179
}
180
}
181
182
if (! ok)
183
{
184
ofstream file(idfilename.c_str());
185
186
file << my_id.str() << endl;
187
file.close();
188
}
189
190
191
TRACE_EXIT();
192
}
193
194
195
//! Returns the id of this node.
196
string
197
DistributionSocketLink::get_my_id() const
198
{
199
return my_id.str();
200
}
201
202
203
//! Returns the total number of peer in the network.
204
int
205
DistributionSocketLink::get_number_of_peers()
206
{
207
int count = 0;
208
209
list<Client *>::iterator i = clients.begin();
210
while (i != clients.end())
211
{
212
Client *c = *i;
213
if (c->socket != NULL)
214
{
215
count++;
216
}
217
i++;
218
}
219
220
return count;
221
}
222
223
224
//! Join the WR network.
225
void
226
DistributionSocketLink::connect(string url)
227
{
228
if (network_enabled)
229
{
230
std::string::size_type pos = url.find("://");
231
std::string hostport;
232
233
if (pos == std::string::npos)
234
{
235
hostport = url;
236
}
237
else
238
{
239
hostport = url.substr(pos + 3);
240
}
241
242
pos = hostport.rfind(":");
243
std::string host;
244
std::string port = "0";
245
246
if (pos == std::string::npos)
247
{
248
host = hostport;
249
}
250
else
251
{
252
host = hostport.substr(0, pos);
253
port = hostport.substr(pos + 1);
254
}
255
256
add_client(NULL, (gchar *)host.c_str(), atoi(port.c_str()), CLIENTTYPE_DIRECT);
257
}
258
}
259
260
261
//! Disconnects the specified client.
262
void
263
DistributionSocketLink::disconnect(string id)
264
{
265
TRACE_ENTER_MSG("DistributionSocketLink::disconnect", id);
266
Client *c = find_client_by_id((gchar*)id.c_str());
267
if (c != NULL)
268
{
269
TRACE_MSG("close");
270
close_client(c);
271
}
272
TRACE_EXIT();
273
}
274
275
276
//! Disconnects all clients.
277
bool
278
DistributionSocketLink::disconnect_all()
279
{
280
list<Client *>::iterator i = clients.begin();
281
bool ret = false;
282
283
master_client = NULL;
284
285
while (i != clients.end())
286
{
287
close_client(*i);
288
ret = true;
289
i++;
290
}
291
292
set_me_master();
293
294
return ret;
295
}
296
297
298
//! Reconnects all clients.
299
bool
300
DistributionSocketLink::reconnect_all()
301
{
302
list<Client *>::iterator i = clients.begin();
303
bool ret = false;
304
305
while (i != clients.end())
306
{
307
if ((*i)->type == CLIENTTYPE_DIRECT)
308
{
309
close_client(*i, true);
310
ret = true;
311
}
312
i++;
313
}
314
315
return ret;
316
}
317
318
319
//! Attempt to become the master node.
320
/*!
321
* \return true if the claim was successfull.
322
*/
323
bool
324
DistributionSocketLink::claim()
325
{
326
TRACE_ENTER("DistributionSocketLink::claim");
327
bool ret = true;
328
329
if (master_client != NULL)
330
{
331
// Another client is master. Politely request to become
332
// master client.
333
send_claim(master_client);
334
ret = false;
335
}
336
else if (!i_am_master && clients.size() > 0)
337
{
338
// No one is master. Just force to be master
339
// potential problem when more client do this simultaneously...
340
send_new_master();
341
i_am_master = true;
342
}
343
else
344
{
345
// No one master, no other clients. Be happy.
346
i_am_master = true;
347
}
348
TRACE_EXIT();
349
return ret;
350
}
351
352
353
//! Lock the master status. Claim will be denied when locked.
354
bool
355
DistributionSocketLink::set_lock_master(bool lock)
356
{
357
master_locked = lock;
358
return true;
359
}
360
361
362
//! Sets the username and password.
363
void
364
DistributionSocketLink::set_user(string user, string pw)
365
{
366
g_free(username);
367
g_free(password);
368
369
username = g_strdup(user.c_str());
370
password = g_strdup(pw.c_str());
371
}
372
373
374
//! Sets the distribution manager for callbacks.
375
void
376
DistributionSocketLink::set_distribution_manager(DistributionManager *dll)
377
{
378
dist_manager = dll;
379
}
380
381
382
//! Enable/disable connecting to distributed operation.
383
bool
384
DistributionSocketLink::set_network_enabled(bool enabled)
385
{
386
if (network_enabled && !enabled)
387
{
388
network_enabled = enabled;
389
set_server_enabled(false);
390
set_me_master();
391
}
392
else if (!network_enabled && enabled)
393
{
394
network_enabled = enabled;
395
396
set_server_enabled(server_enabled);
397
}
398
399
return network_enabled;
400
}
401
402
403
//! Enable/disable listening for distributed operation.
404
bool
405
DistributionSocketLink::set_server_enabled(bool enabled)
406
{
407
TRACE_ENTER_MSG("DistributionSocketLink:set_server_enabled", enabled);
408
bool ret = server_enabled;
409
410
if (!network_enabled) {
411
// Don't start the server if the network is not enabled
412
return ret;
413
}
414
415
if (server_socket == NULL && enabled)
416
{
417
// Switching from disabled to enabled;
418
if (!start_async_server())
419
{
420
// We did not succeed in starting the server. Arghh.
421
dist_manager->log(_("Could not enable network operation."));
422
enabled = false;
423
}
424
}
425
else if (server_socket != NULL && !enabled)
426
{
427
// Switching from enabled to disabled.
428
if (server_socket != NULL)
429
{
430
dist_manager->log(_("Disabling network operation."));
431
432
delete server_socket;
433
}
434
server_socket = NULL;
435
disconnect_all();
436
}
437
438
server_enabled = enabled;
439
TRACE_EXIT();
440
return ret;
441
}
442
443
444
//! Register a distributed client message callback.
445
bool
446
DistributionSocketLink::register_client_message(DistributionClientMessageID id,
447
DistributionClientMessageType type,
448
IDistributionClientMessage *callback)
449
{
450
ClientMessageListener sl;
451
sl.listener = callback;
452
sl.type = type;
453
454
client_message_map[id] = sl;
455
456
return true;
457
}
458
459
460
//! Unregister a distributed clien _message callback.
461
bool
462
DistributionSocketLink::unregister_client_message(DistributionClientMessageID id)
463
{
464
(void) id;
465
return false;
466
}
467
468
469
//! Force is state distribution.
470
bool
471
DistributionSocketLink::broadcast_client_message(DistributionClientMessageID dsid,
472
PacketBuffer &buffer)
473
{
474
TRACE_ENTER("DistributionSocketLink::broadcast_client_message");
475
476
PacketBuffer packet;
477
packet.create();
478
init_packet(packet, PACKET_CLIENTMSG);
479
480
string id = get_master();
481
packet.pack_string(id);
482
483
packet.pack_ushort(1);
484
packet.pack_ushort(dsid);
485
packet.pack_ushort(buffer.bytes_written());
486
packet.pack_raw((unsigned char *)buffer.get_buffer(),
487
buffer.bytes_written());
488
489
send_packet_broadcast(packet);
490
TRACE_EXIT();
491
return true;
492
}
493
494
495
//! Returns whether the specified client is this client.
496
bool
497
DistributionSocketLink::client_is_me(gchar *id)
498
{
499
return id != NULL && strcmp(id, my_id.str().c_str()) == 0;
500
}
501
502
503
504
//! Returns whether the specified client exists.
505
/*!
506
* This method checks if the specified client is an existing remote
507
* client or the local client.
508
*/
509
bool
510
DistributionSocketLink::exists_client(gchar *id)
511
{
512
TRACE_ENTER_MSG("DistributionSocketLink::exists_client", id);
513
514
bool ret = client_is_me(id);
515
516
if (!ret)
517
{
518
Client *c = find_client_by_id(id);
519
ret = (c != NULL);
520
}
521
522
TRACE_EXIT();
523
return ret;
524
}
525
526
527
//! Adds a new client and connect to it.
528
bool
529
DistributionSocketLink::add_client(gchar *id, gchar *host, gint port, ClientType type, Client *peer)
530
{
531
TRACE_ENTER_MSG("DistributionSocketLink::add_client",
532
(id != NULL ? id : "NULL") << " " <<
533
(host != NULL ? host : "NULL") << " " << port);
534
535
gchar *canonical_host = NULL;
536
537
Client *c = find_client_by_canonicalname(host, port);
538
if (c != NULL && c->type == CLIENTTYPE_SIGNEDOFF && type == CLIENTTYPE_DIRECT)
539
{
540
if (c->id != NULL)
541
{
542
dist_manager->signon_remote_client(c->id);
543
}
544
545
c->type = type;
546
dist_manager->log(_("Connecting to %s."), host);
547
548
if (c->socket != NULL)
549
{
550
c->socket->close();
551
delete c->socket;
552
}
553
554
ISocket *socket = socket_driver->create_socket();
555
socket->set_data(c);
556
socket->set_listener(this);
557
socket->connect(host, port);
558
559
c->socket = socket;
560
}
561
else
562
{
563
// Client does not yet exists as far as we can see.
564
// So, create a new one.
565
Client *client = new Client;
566
567
client->type = type;
568
client->peer = peer;
569
client->packet.create();
570
client->hostname = g_strdup(host);
571
client->id = g_strdup(id);
572
client->port = port;
573
574
clients.push_back(client);
575
576
if (client->id != NULL)
577
{
578
dist_manager->signon_remote_client(client->id);
579
}
580
581
if (type == CLIENTTYPE_DIRECT)
582
{
583
dist_manager->log(_("Connecting to %s."), host);
584
585
if (client->socket != NULL)
586
{
587
client->socket->close();
588
delete client->socket;
589
}
590
591
ISocket *socket = socket_driver->create_socket();
592
socket->set_data(client);
593
socket->set_listener(this);
594
socket->connect(host, port);
595
596
client->socket = socket;
597
}
598
}
599
g_free(canonical_host);
600
TRACE_EXIT();
601
return true;
602
}
603
604
605
//! Sets the id of a client.
606
/*!
607
* This method also checks for duplicates.
608
*
609
* \return true if the client is a duplicate. The id will not
610
* changed if the client is a duplicate.
611
*/
612
bool
613
DistributionSocketLink::set_client_id(Client *client, gchar *id)
614
{
615
TRACE_ENTER_MSG("DistributionSocketLink::set_id", id);
616
bool ret = true;
617
618
bool exists = exists_client(id);
619
if (exists)
620
{
621
// Already have a client with this name/port
622
Client *old_client = find_client_by_id(id);
623
624
if (old_client == NULL)
625
{
626
// Iek this is me.
627
TRACE_MSG("It'me me");
628
ret = false;
629
}
630
else if (old_client != client)
631
{
632
TRACE_MSG("It's not me " << old_client->type << " " << old_client->socket);
633
// It's a remote client, but not the same one.
634
635
bool reuse = ((old_client->type == CLIENTTYPE_DIRECT ||
636
old_client->type == CLIENTTYPE_SIGNEDOFF)
637
&& old_client->socket == NULL);
638
639
TRACE_MSG("reuse " << reuse);
640
if (reuse)
641
{
642
// Client exist, but is not connected.
643
// Silently remove the old client.
644
remove_client(old_client);
645
}
646
else
647
{
648
// Already connected to this client.
649
// Duplicate.
650
ret = false;
651
}
652
}
653
else
654
{
655
// it's ok. It's same one.
656
}
657
}
658
659
if (ret)
660
{
661
// No duplicate, so change the canonical name.
662
g_free(client->id);
663
g_free(client->hostname);
664
client->id = g_strdup(id);
665
client->hostname = NULL;
666
client->port = 0;
667
668
if (client->id != NULL)
669
{
670
dist_manager->signon_remote_client(client->id);
671
}
672
}
673
674
TRACE_RETURN(ret);
675
return ret;
676
}
677
678
679
//! Removes a client (or all clients)
680
/*!
681
* Network connections to removed client are closed.
682
*
683
* \param client client to remove, or NULL if all clients to be removed.
684
*
685
*/
686
void
687
DistributionSocketLink::remove_client(Client *client)
688
{
689
list<Client *>::iterator i = clients.begin();
690
691
if (client == master_client)
692
{
693
// Client to be removed is master. Unset master client.
694
master_client = NULL;
695
}
696
697
while (i != clients.end())
698
{
699
if (client == NULL || *i == client || (*i)->peer == client)
700
{
701
dist_manager->log(_("Removing client %s."),
702
(*i)->id == NULL ? "Unknown" : (*i)->id);
703
delete *i;
704
i = clients.erase(i);
705
}
706
else
707
{
708
i++;
709
}
710
}
711
}
712
713
714
//! Removes all peers of the specified client.
715
void
716
DistributionSocketLink::remove_peer_clients(Client *client)
717
{
718
TRACE_ENTER("DistributionSocketLink::remove_peer_clients");
719
720
list<Client *>::iterator i = clients.begin();
721
while (i != clients.end())
722
{
723
if ((*i)->peer == client)
724
{
725
TRACE_MSG("Client " << (*i)->peer->id << " is peer of " << client->id);
726
727
dist_manager->log(_("Removing client %s."),
728
(*i)->id == NULL ? "Unknown" : (*i)->id);
729
send_signoff(NULL, *i);
730
731
if (client == master_client)
732
{
733
// Connection to master is lost. Unset master client.
734
set_master(NULL);
735
}
736
737
delete *i;
738
i = clients.erase(i);
739
}
740
else
741
{
742
i++;
743
}
744
}
745
TRACE_EXIT();
746
}
747
748
749
//! Closes the connection to a client.
750
void
751
DistributionSocketLink::close_client(Client *client, bool reconnect /* = false*/)
752
{
753
TRACE_ENTER_MSG("DistributionSocketLink::close_client",
754
(client->id != NULL ? client->id : "Unknown") << " " << reconnect);
755
756
if (client == master_client)
757
{
758
// Client to be closed is master. Unset master client.
759
set_master(NULL);
760
}
761
762
if (client->type == CLIENTTYPE_DIRECT)
763
{
764
TRACE_MSG("Is direct");
765
// Closing direct connection.
766
dist_manager->log(_("Disconnecting %s"),
767
client->id != NULL ? client->id : "Unknown");
768
769
// Inform the client that we are disconnecting.
770
send_signoff(client, NULL);
771
772
if (client->socket != NULL)
773
{
774
TRACE_MSG("Still connected");
775
// Still connected. Disconect.
776
delete client->socket;
777
client->socket = NULL;
778
779
if (reconnect)
780
{
781
TRACE_MSG("must reconnected");
782
client->reconnect_count = reconnect_attempts;
783
client->reconnect_time = time(NULL) + 5;
784
}
785
else
786
{
787
TRACE_MSG("set signed off");
788
client->reconnect_count = 0;
789
client->reconnect_time = 0;
790
client->type = CLIENTTYPE_SIGNEDOFF;
791
}
792
}
793
794
send_signoff(NULL, client);
795
remove_peer_clients(client);
796
}
797
else if (client->type == CLIENTTYPE_SIGNEDOFF)
798
{
799
TRACE_MSG("is signed off");
800
if (client->socket != NULL)
801
{
802
TRACE_MSG("still connected");
803
804
// Still connected. Disconect.
805
delete client->socket;
806
client->socket = NULL;
807
808
client->reconnect_count = 0;
809
client->reconnect_time = 0;
810
}
811
812
remove_peer_clients(client);
813
}
814
}
815
816
817
//! Check if a client point is stil valid....
818
bool
819
DistributionSocketLink::is_client_valid(Client *client)
820
{
821
list<Client *>::iterator i = clients.begin();
822
bool ret = false;
823
824
while (!ret && i != clients.end())
825
{
826
if (*i == client)
827
{
828
ret = true;
829
}
830
i++;
831
}
832
833
return ret;
834
}
835
836
837
//! Finds a remote client by its canonical name and port.
838
DistributionSocketLink::Client *
839
DistributionSocketLink::find_client_by_canonicalname(gchar *name, gint port)
840
{
841
Client *ret = NULL;
842
list<Client *>::iterator i = clients.begin();
843
844
while (i != clients.end())
845
{
846
if ((*i)->port == port && (*i)->hostname != NULL && strcmp((*i)->hostname, name) == 0)
847
{
848
ret = *i;
849
}
850
i++;
851
}
852
return ret;
853
}
854
855
//! Finds a remote client by its id.
856
DistributionSocketLink::Client *
857
DistributionSocketLink::find_client_by_id(gchar *id)
858
{
859
Client *ret = NULL;
860
list<Client *>::iterator i = clients.begin();
861
862
while (i != clients.end())
863
{
864
if ((*i)->id != NULL && strcmp((*i)->id, id) == 0)
865
{
866
ret = *i;
867
}
868
i++;
869
}
870
return ret;
871
}
872
873
874
//! Returns the master client.
875
string
876
DistributionSocketLink::get_master() const
877
{
878
string id;
879
880
if (i_am_master)
881
{
882
id = get_my_id();
883
}
884
else if (master_client != NULL && master_client->id != NULL)
885
{
886
id = master_client->id;
887
}
888
return id;
889
}
890
891
892
//! Sets the specified remote client as master.
893
void
894
DistributionSocketLink::set_master(Client *client)
895
{
896
TRACE_ENTER("DistributionSocketLink::set_master")
897
master_client = client;
898
i_am_master = false;
899
900
if (dist_manager != NULL)
901
{
902
dist_manager->master_changed(false, client != NULL ? client->id : "");
903
}
904
TRACE_EXIT();
905
}
906
907
908
//! Sets the local client as master.
909
void
910
DistributionSocketLink::set_me_master()
911
{
912
TRACE_ENTER("DistributionSocketLink::set_me_master");
913
master_client = NULL;
914
i_am_master = true;
915
916
if (dist_manager != NULL)
917
{
918
dist_manager->master_changed(true, get_my_id());
919
}
920
TRACE_EXIT();
921
}
922
923
924
//! Sets the specified client master.
925
void
926
DistributionSocketLink::set_master_by_id(gchar *id)
927
{
928
TRACE_ENTER_MSG("DistributionSocketLink::set_master", id);
929
930
Client *c = find_client_by_id(id);
931
932
if (c != NULL)
933
{
934
// It's a remote client. mark it master.
935
dist_manager->log(_("Client %s is now master."),
936
c->id == NULL ? "Unknown" : c->id);
937
set_master(c);
938
}
939
else if (strcmp(id, get_my_id().c_str()) == 0)
940
{
941
// Its ME!
942
dist_manager->log(_("I'm now master."));
943
set_me_master();
944
}
945
else
946
{
947
// Huh???
948
TRACE_MSG("Iek");
949
}
950
951
TRACE_EXIT();
952
}
953
954
955
//! Initialize an outgoing packet.
956
void
957
DistributionSocketLink::init_packet(PacketBuffer &packet, PacketCommand cmd)
958
{
959
// Length.
960
packet.pack_ushort(0);
961
// Version
962
packet.pack_byte(2);
963
// Flags
964
packet.pack_byte(0);
965
// Command
966
packet.pack_ushort(cmd);
967
}
968
969
970
//! Sends the specified packet to all clients.
971
void
972
DistributionSocketLink::send_packet_broadcast(PacketBuffer &packet)
973
{
974
TRACE_ENTER("DistributionSocketLink::send_packet_broadcast");
975
976
send_packet_except(packet, NULL);
977
978
TRACE_EXIT();
979
}
980
981
982
//! Sends the specified packet to all clients with the exception of one client.
983
void
984
DistributionSocketLink::send_packet_except(PacketBuffer &packet, Client *client)
985
{
986
TRACE_ENTER("DistributionSocketLink::send_packet_except");
987
988
gint size = packet.bytes_written();
989
990
// Length.
991
packet.poke_ushort(0, size);
992
993
list<Client *>::iterator i = clients.begin();
994
while (i != clients.end())
995
{
996
Client *c = *i;
997
998
if (c != client && c->socket != NULL)
999
{
1000
int bytes_written = 0;
1001
c->socket->write(packet.get_buffer(), size, bytes_written);
1002
}
1003
i++;
1004
}
1005
1006
TRACE_EXIT();
1007
}
1008
1009
1010
//! Sends the specified packet to the specified client.
1011
void
1012
DistributionSocketLink::send_packet(Client *client, PacketBuffer &packet)
1013
{
1014
TRACE_ENTER("DistributionSocketLink::send_packet");
1015
1016
if (client != NULL && client->type == CLIENTTYPE_ROUTED)
1017
{
1018
TRACE_MSG("Must route packet.");
1019
1020
if (client->id == NULL)
1021
{
1022
TRACE_MSG("Client's ID == NULL");
1023
}
1024
1025
packet.restart_read();
1026
int flags = packet.peek_byte(3);
1027
1028
if (!(flags & PACKETFLAG_DEST) && client->id != NULL)
1029
{
1030
assert(!(flags & PACKETFLAG_SOURCE));
1031
1032
TRACE_MSG("Add destination " << client->id);
1033
1034
packet.poke_byte(3, flags | PACKETFLAG_DEST);
1035
1036
packet.insert(4, strlen(client->id) + 2);
1037
packet.poke_string(6, client->id);
1038
}
1039
client = client->peer;
1040
}
1041
1042
if (client != NULL && client->socket != NULL)
1043
{
1044
if (client->id != NULL)
1045
{
1046
TRACE_MSG("Sending to " << client->id);
1047
}
1048
1049
gint size = packet.bytes_written();
1050
1051
// Length.
1052
packet.poke_ushort(0, size);
1053
1054
int bytes_written = 0;
1055
client->socket->write(packet.get_buffer(), size, bytes_written);
1056
}
1057
1058
TRACE_EXIT();
1059
}
1060
1061
1062
//! Processed an incoming packet.
1063
void
1064
DistributionSocketLink::process_client_packet(Client *client)
1065
{
1066
TRACE_ENTER("DistributionSocketLink::process_client_packet");
1067
PacketBuffer &packet = client->packet;
1068
1069
client->claim_count = 0;
1070
1071
gint size = packet.unpack_ushort();
1072
g_assert(size == packet.bytes_written());
1073
1074
gint version = packet.unpack_byte();
1075
gint flags = packet.unpack_byte();
1076
1077
gint type = packet.unpack_ushort();
1078
TRACE_MSG("type = " << type);
1079
if (client != NULL && client->id != NULL)
1080
{
1081
TRACE_MSG("From = " << client->id);
1082
}
1083
1084
Client *source = client;
1085
bool forward = true;
1086
if (flags & PACKETFLAG_SOURCE)
1087
{
1088
gchar *id = packet.unpack_string();
1089
1090
if (!client_is_me(id))
1091
{
1092
TRACE_MSG("routed, source = " << id);
1093
1094
source = find_client_by_id(id);
1095
1096
if (source == NULL)
1097
{
1098
TRACE_MSG("Unknown source. Dropping");
1099
}
1100
else if (source != client && source->peer != client)
1101
{
1102
TRACE_MSG("Illegal source in routing.");
1103
source = NULL;
1104
}
1105
}
1106
else
1107
{
1108
TRACE_MSG("Cycle detected.");
1109
type = 0;
1110
flags = 0;
1111
source = NULL;
1112
1113
// Duplicate client. inform client that it's bogus and close.
1114
// dist_manager->log(_("Client %s:%d is duplicate."), client->hostname, client->port);
1115
1116
// send_duplicate(client);
1117
// remove_client(client);
1118
}
1119
g_free(id);
1120
}
1121
1122
if (flags & PACKETFLAG_DEST)
1123
{
1124
gchar *id = packet.unpack_string();
1125
1126
if (id != NULL && !client_is_me(id))
1127
{
1128
TRACE_MSG("Destination = " << id);
1129
Client *dest = find_client_by_id(id);
1130
1131
if (dest != NULL)
1132
{
1133
TRACE_MSG("Forwarding to destination");
1134
forward_packet(packet, dest, source);
1135
}
1136
1137
source = NULL;
1138
}
1139
g_free(id);
1140
}
1141
1142
TRACE_MSG("size = " << size << ", version = " << version << ", flags = " << flags);
1143
1144
if (source != NULL || type == PACKET_CLIENT_LIST)
1145
{
1146
switch (type)
1147
{
1148
case PACKET_HELLO:
1149
handle_hello(packet, source);
1150
forward = false;
1151
break;
1152
1153
case PACKET_SIGNOFF:
1154
handle_signoff(packet, source);
1155
break;
1156
1157
case PACKET_CLAIM:
1158
handle_claim(packet, source);
1159
break;
1160
1161
case PACKET_WELCOME:
1162
handle_welcome(packet, source);
1163
forward = false;
1164
break;
1165
1166
case PACKET_CLIENT_LIST:
1167
forward = handle_client_list(packet, source, client);
1168
break;
1169
1170
case PACKET_NEW_MASTER:
1171
handle_new_master(packet, source);
1172
break;
1173
1174
case PACKET_CLIENTMSG:
1175
handle_client_message(packet, source);
1176
break;
1177
1178
case PACKET_DUPLICATE:
1179
handle_duplicate(packet, source);
1180
forward = false;
1181
break;
1182
1183
case PACKET_CLAIM_REJECT:
1184
handle_claim_reject(packet, source);
1185
break;
1186
}
1187
1188
if (forward)
1189
{
1190
forward_packet_except(packet, client, source);
1191
}
1192
}
1193
1194
packet.clear();
1195
packet.resize(0);
1196
TRACE_EXIT();
1197
}
1198
1199
1200
void
1201
DistributionSocketLink::forward_packet_except(PacketBuffer &packet, Client *client, Client *source)
1202
{
1203
TRACE_ENTER("DistributionSocketLink::forward_packet_except");
1204
1205
packet.restart_read();
1206
int flags = packet.peek_byte(3);
1207
if (!(flags & PACKETFLAG_SOURCE) && source->id != NULL)
1208
{
1209
TRACE_MSG("Add source " << source->id);
1210
packet.poke_byte(3, flags | PACKETFLAG_SOURCE);
1211
packet.insert(4, strlen(source->id) + 2);
1212
packet.poke_string(6, source->id);
1213
}
1214
send_packet_except(packet, client);
1215
1216
TRACE_EXIT();
1217
}
1218
1219
1220
void
1221
DistributionSocketLink::forward_packet(PacketBuffer &packet, Client *dest, Client *source)
1222
{
1223
TRACE_ENTER("DistributionSocketLink::forward_packet");
1224
1225
packet.restart_read();
1226
int flags = packet.peek_byte(3);
1227
if (!(flags & PACKETFLAG_SOURCE) && source->id != NULL)
1228
{
1229
TRACE_MSG("Add source " << source->id);
1230
packet.poke_byte(3, flags | PACKETFLAG_SOURCE);
1231
packet.insert(4, strlen(source->id) + 2);
1232
packet.poke_string(6, source->id);
1233
}
1234
send_packet(dest, packet);
1235
TRACE_EXIT();
1236
}
1237
1238
1239
//! Sends a hello to the specified client.
1240
void
1241
DistributionSocketLink::send_hello(Client *client)
1242
{
1243
TRACE_ENTER("DistributionSocketLink::send_hello");
1244
1245
PacketBuffer packet;
1246
1247
packet.create();
1248
init_packet(packet, PACKET_HELLO);
1249
1250
packet.pack_string(username);
1251
packet.pack_string(password);
1252
packet.pack_string(get_my_id());
1253
packet.pack_string(get_my_id()); // was: hostname
1254
packet.pack_ushort(server_port);
1255
1256
send_packet(client, packet);
1257
TRACE_EXIT();
1258
}
1259
1260
1261
//! Handles a Hello from the specified client.
1262
void
1263
DistributionSocketLink::handle_hello(PacketBuffer &packet, Client *client)
1264
{
1265
TRACE_ENTER("DistributionSocketLink::handle_hello");
1266
1267
gchar *user = packet.unpack_string();
1268
gchar *pass = packet.unpack_string();
1269
gchar *id = packet.unpack_string();
1270
/* gchar *name = */ packet.unpack_string();
1271
/* int port = */ packet.unpack_ushort();
1272
1273
dist_manager->log(_("Client %s saying hello."), id != NULL ? id : "Unknown");
1274
1275
if ( (username == NULL || (user != NULL && strcmp(username, user) == 0)) &&
1276
(password == NULL || (pass != NULL && strcmp(password, pass) == 0)))
1277
{
1278
bool ok = set_client_id(client, id);
1279
1280
if (ok)
1281
{
1282
// Welcome!
1283
send_welcome(client);
1284
}
1285
else
1286
{
1287
// Duplicate client. inform client that it's bogus and close.
1288
dist_manager->log(_("Client %s is duplicate."),
1289
id != NULL ? id : "Unknown");
1290
1291
send_duplicate(client);
1292
remove_client(client);
1293
}
1294
}
1295
else
1296
{
1297
// Incorrect password.
1298
dist_manager->log(_("Client %s access denied."),
1299
id != NULL ? id : "Unknown");
1300
remove_client(client);
1301
}
1302
1303
g_free(user);
1304
g_free(id);
1305
g_free(pass);
1306
1307
TRACE_EXIT();
1308
}
1309
1310
1311
1312
//! Sends a hello to the specified client.
1313
void
1314
DistributionSocketLink::send_signoff(Client *to, Client *signedoff_client)
1315
{
1316
TRACE_ENTER("DistributionSocketLink::send_signoff");
1317
1318
PacketBuffer packet;
1319
1320
packet.create();
1321
init_packet(packet, PACKET_SIGNOFF);
1322
1323
if (signedoff_client != NULL)
1324
{
1325
TRACE_MSG("remote client " << (signedoff_client->id != NULL ? signedoff_client->id : "?"));
1326
packet.pack_string(signedoff_client->id);
1327
}
1328
else
1329
{
1330
TRACE_MSG("me " << my_id.str());
1331
packet.pack_string(get_my_id());
1332
}
1333
1334
if (to != NULL)
1335
{
1336
TRACE_MSG("sending to " << (to->id != NULL ? to->id : "?"));
1337
send_packet(to, packet);
1338
}
1339
else
1340
{
1341
TRACE_MSG("broadcasting");
1342
send_packet_broadcast(packet);
1343
}
1344
TRACE_EXIT();
1345
}
1346
1347
1348
//! Handles a Hello from the specified client.
1349
void
1350
DistributionSocketLink::handle_signoff(PacketBuffer &packet, Client *client)
1351
{
1352
TRACE_ENTER("DistributionSocketLink::handle_signoff");
1353
1354
(void) client;
1355
1356
gchar *id = packet.unpack_string();
1357
Client *c = NULL;
1358
1359
if (id != NULL)
1360
{
1361
c = find_client_by_id(id);
1362
g_free(id);
1363
}
1364
1365
if (c != NULL)
1366
{
1367
dist_manager->log(_("Client %s signed off."),
1368
c->id == NULL ? "Unknown" : c->id);
1369
1370
if (c->type == CLIENTTYPE_DIRECT)
1371
{
1372
TRACE_MSG("Direct connection. setting signedoff");
1373
c->type = CLIENTTYPE_SIGNEDOFF;
1374
remove_peer_clients(c);
1375
1376
if (c->socket != NULL)
1377
{
1378
TRACE_MSG("Remove connection");
1379
delete c->socket;
1380
c->socket = NULL;
1381
}
1382
1383
remove_client(c);
1384
}
1385
else
1386
{
1387
TRACE_MSG("Routed connection. removing");
1388
remove_client(c);
1389
}
1390
}
1391
1392
TRACE_EXIT();
1393
}
1394
1395
1396
//! Sends a duplicate to the specified client.
1397
void
1398
DistributionSocketLink::send_duplicate(Client *client)
1399
{
1400
TRACE_ENTER("DistributionSocketLink::send_duplicate");
1401
1402
PacketBuffer packet;
1403
1404
packet.create();
1405
init_packet(packet, PACKET_DUPLICATE);
1406
1407
send_packet(client, packet);
1408
TRACE_EXIT();
1409
}
1410
1411
1412
//! Handles a duplicate for the specified client.
1413
void
1414
DistributionSocketLink::handle_duplicate(PacketBuffer &packet, Client *client)
1415
{
1416
(void) packet;
1417
TRACE_ENTER("DistributionSocketLink::handle_duplicate");
1418
dist_manager->log(_("Client %s is duplicate."),
1419
client->id == NULL ? "Unknown" : client->id);
1420
remove_client(client);
1421
1422
TRACE_EXIT();
1423
}
1424
1425
1426
//! Sends a welcome message to the specified client
1427
void
1428
DistributionSocketLink::send_welcome(Client *client)
1429
{
1430
TRACE_ENTER("DistributionSocketLink::send_welcome");
1431
1432
PacketBuffer packet;
1433
1434
packet.create();
1435
init_packet(packet, PACKET_WELCOME);
1436
1437
// My Info
1438
packet.pack_string(get_my_id());
1439
packet.pack_string(get_my_id()); // was: hostname
1440
packet.pack_ushort(server_port);
1441
1442
send_packet(client, packet);
1443
TRACE_EXIT();
1444
}
1445
1446
1447
//! Handles a welcome message from the specified client.
1448
void
1449
DistributionSocketLink::handle_welcome(PacketBuffer &packet, Client *client)
1450
{
1451
TRACE_ENTER("DistributionSocketLink::handle_welcome");
1452
1453
gchar *id = packet.unpack_string();
1454
gchar *name = packet.unpack_string();
1455
/*gint port = */ packet.unpack_ushort();
1456
1457
dist_manager->log(_("Client %s is welcoming us."),
1458
id == NULL ? "Unknown" : id);
1459
1460
bool ok = set_client_id(client, id);
1461
1462
if (ok)
1463
{
1464
// The connected client offers the master client.
1465
// This info will be received in the client list.
1466
// So, we no longer know who's master...
1467
set_master(NULL);
1468
1469
// All, ok. Send list of known client.
1470
// WITHOUT info about who's master on out side.
1471
send_client_list(client);
1472
}
1473
else
1474
{
1475
// Duplicate.
1476
send_duplicate(client);
1477
remove_client(client);
1478
}
1479
1480
g_free(id);
1481
g_free(name);
1482
1483
TRACE_EXIT();
1484
}
1485
1486
1487
//! Sends the list of known clients to the specified client.
1488
void
1489
DistributionSocketLink::send_client_list(Client *client, bool except)
1490
{
1491
TRACE_ENTER("DistributionSocketLink::send_client_list");
1492
1493
if (clients.size() > 0)
1494
{
1495
PacketBuffer packet;
1496
packet.create();
1497
init_packet(packet, PACKET_CLIENT_LIST);
1498
1499
int count = 1;
1500
gint clients_pos = packet.bytes_written();
1501
1502
packet.pack_ushort(0); // number of clients in the list
1503
packet.pack_ushort(0); // flags.
1504
1505
// Put muself in list.
1506
gint pos = packet.bytes_written();
1507
1508
TRACE_MSG("client me: " << my_id.str() << " " << server_port << " " << i_am_master);
1509
int flags = CLIENTLIST_ME | (i_am_master ? CLIENTLIST_MASTER : 0);
1510
packet.pack_ushort(0); // Length
1511
packet.pack_ushort(flags); // Flags
1512
packet.pack_string(get_my_id()); // ID
1513
packet.pack_string(get_my_id()); // Canonical name
1514
packet.pack_ushort(server_port); // Listen port.
1515
1516
// Size of the client data.
1517
packet.poke_ushort(pos, packet.bytes_written() - pos);
1518
1519
// Put known client in the list.
1520
list<Client *>::iterator i = clients.begin();
1521
while (i != clients.end())
1522
{
1523
Client *c = *i;
1524
1525
if (c->id != NULL)
1526
{
1527
count++;
1528
pos = packet.bytes_written();
1529
1530
int flags = 0;
1531
if (c == master_client)
1532
{
1533
flags |= CLIENTLIST_MASTER;
1534
}
1535
1536
TRACE_MSG("Send client: " << c->id);
1537
1538
packet.pack_ushort(0); // Length
1539
packet.pack_ushort(flags); // Flags
1540
packet.pack_string(c->id); // ID
1541
packet.pack_string(c->hostname); // Canonical name
1542
packet.pack_ushort(c->port); // Listen port.
1543
1544
// Size of the client data.
1545
packet.poke_ushort(pos, packet.bytes_written() - pos);
1546
}
1547
i++;
1548
}
1549
1550
// Put packet size in the packet and send.
1551
packet.poke_ushort(clients_pos, count);
1552
1553
if (except)
1554
{
1555
send_packet_except(packet, client);
1556
}
1557
else
1558
{
1559
send_packet(client, packet);
1560
}
1561
}
1562
1563
TRACE_EXIT();
1564
}
1565
1566
1567
//! Handles a client list from the specified client.
1568
bool
1569
DistributionSocketLink::handle_client_list(PacketBuffer &packet, Client *client, Client *direct)
1570
{
1571
TRACE_ENTER("DistributionSocketLink::handle_client_list");
1572
1573
// Extract data.
1574
gint num_clients = packet.unpack_ushort();
1575
gint flags = packet.unpack_ushort();
1576
(void) flags;
1577
1578
gchar *master_id = NULL;
1579
1580
gchar **names = new gchar*[num_clients];
1581
gchar **ids = new gchar*[num_clients];
1582
gint *ports = new gint[num_clients];
1583
1584
bool ok = true;
1585
1586
// Loop over remote clients.
1587
for (int i = 0; i < num_clients; i++)
1588
{
1589
names[i] = NULL;
1590
ids[i] = NULL;
1591
ports[i] = 0;
1592
1593
// Extract data.
1594
gint pos = packet.bytes_read();
1595
gint size = packet.unpack_ushort();
1596
gint flags = packet.unpack_ushort();
1597
gchar *id = packet.unpack_string();
1598
gchar *name = packet.unpack_string();
1599
gint port = packet.unpack_ushort();
1600
1601
if (flags & CLIENTLIST_MASTER)
1602
{
1603
master_id = g_strdup(id);
1604
TRACE_MSG("Master: " << master_id);
1605
}
1606
1607
if (id != NULL)
1608
{
1609
if (!exists_client(id))
1610
{
1611
// A new client
1612
TRACE_MSG("new client: " << id);
1613
names[i] = name;
1614
ids[i] = id;
1615
ports[i] = port;
1616
}
1617
else if (client != NULL && direct == client && !client_is_me(id) && strcmp(client->id, id) != 0)
1618
{
1619
TRACE_MSG("Strange client: " << id);
1620
ok = false;
1621
}
1622
}
1623
1624
// Skip trailing junk...
1625
size -= (packet.bytes_read() - pos);
1626
packet.skip(size);
1627
1628
g_free(name);
1629
g_free(id);
1630
}
1631
1632
1633
if (ok)
1634
{
1635
// And send the list of client we are connected to.
1636
if (client != NULL && direct == client && !client->sent_client_list)
1637
{
1638
client->sent_client_list = true;
1639
send_client_list(client);
1640
}
1641
1642
TRACE_MSG("Adding: ");
1643
for (int i = 0; i < num_clients; i++)
1644
{
1645
if (ids[i] != NULL && names[i] != NULL)
1646
{
1647
add_client(ids[i], names[i], ports[i], CLIENTTYPE_ROUTED, direct);
1648
}
1649
}
1650
1651
if (master_id != NULL)
1652
{
1653
set_master_by_id(master_id);
1654
TRACE_MSG(master_id << " is now master");
1655
}
1656
1657
send_client_message(DCMT_SIGNON);
1658
}
1659
else
1660
{
1661
TRACE_MSG("Dup: ");
1662
dist_manager->log(_("Client %s is duplicate."),
1663
client->id != NULL ? client->id : "Unknown");
1664
1665
send_duplicate(client);
1666
remove_client(client);
1667
}
1668
1669
TRACE_MSG("Ok: ");
1670
1671
1672
for (int i = 0; i < num_clients; i++)
1673
{
1674
g_free(ids[i]);
1675
g_free(names[i]);
1676
}
1677
1678
g_free(master_id);
1679
delete [] names;
1680
delete [] ids;
1681
delete [] ports;
1682
1683
TRACE_EXIT();
1684
return ok;
1685
}
1686
1687
1688
//! Requests to become master.
1689
void
1690
DistributionSocketLink::send_claim(Client *client)
1691
{
1692
TRACE_ENTER("DistributionSocketLink::send_claim");
1693
1694
if (client->next_claim_time == 0 || time(NULL) >= client->next_claim_time)
1695
{
1696
PacketBuffer packet;
1697
1698
dist_manager->log(_("Requesting master status from %s."),
1699
client->id == NULL ? "Unknown" : client->id);
1700
1701
packet.create();
1702
init_packet(packet, PACKET_CLAIM);
1703
1704
packet.pack_ushort(0);
1705
1706
client->next_claim_time = time(NULL) + 10;
1707
1708
send_packet(client, packet);
1709
1710
if (client->claim_count >= 3)
1711
{
1712
dist_manager->log(_("Client timeout from %s."),
1713
client->id == NULL ? "Unknown" : client->id);
1714
1715
close_client(client, client->outbound);
1716
}
1717
client->claim_count++;
1718
}
1719
1720
TRACE_EXIT();
1721
}
1722
1723
1724
//! Handles a request from a remote client to become master.
1725
void
1726
DistributionSocketLink::handle_claim(PacketBuffer &packet, Client *client)
1727
{
1728
TRACE_ENTER("DistributionSocketLink::handle_claim");
1729
1730
/*gint count = */ packet.unpack_ushort();
1731
1732
if (i_am_master && master_locked)
1733
{
1734
dist_manager->log(_("Rejecting master request from client %s."),
1735
client->id == NULL ? "Unknown" : client->id);
1736
send_claim_reject(client);
1737
}
1738
else
1739
{
1740
dist_manager->log(_("Acknowledging master request from client %s."),
1741
client->id == NULL ? "Unknown" : client->id);
1742
1743
bool was_master = i_am_master;
1744
1745
// Marks client as master
1746
set_master(client);
1747
assert(!i_am_master);
1748
1749
// If I was previously master, distribute state.
1750
if (was_master)
1751
{
1752
//dist_manager->log(_("Transferring state to client %s:%d."),
1753
// client->hostname, client->port);
1754
send_client_message(DCMT_MASTER);
1755
}
1756
1757
// And tell everyone we have a new master.
1758
send_new_master();
1759
}
1760
1761
TRACE_EXIT();
1762
}
1763
1764
1765
1766
//! Inform that the claim has been rejected.
1767
void
1768
DistributionSocketLink::send_claim_reject(Client *client)
1769
{
1770
TRACE_ENTER("DistributionSocketLink::send_claim_reject");
1771
1772
PacketBuffer packet;
1773
1774
packet.create();
1775
init_packet(packet, PACKET_CLAIM_REJECT);
1776
1777
send_packet(client, packet);
1778
TRACE_EXIT();
1779
}
1780
1781
1782
//! Handles a rejection of my claim.
1783
void
1784
DistributionSocketLink::handle_claim_reject(PacketBuffer &packet, Client *client)
1785
{
1786
TRACE_ENTER("DistributionSocketLink::handle_claim");
1787
(void) packet;
1788
1789
if (client != master_client)
1790
{
1791
dist_manager->log(_("Non-master client %s rejected master request."),
1792
client->id == NULL ? "Unknown" : client->id);
1793
}
1794
else
1795
{
1796
dist_manager->log(_("Client %s rejected master request, delaying."),
1797
client->id == NULL ? "Unknown" : client->id);
1798
client->reject_count++;
1799
int count = client->reject_count;
1800
1801
if (count > 6)
1802
{
1803
count = 6;
1804
}
1805
1806
client->next_claim_time = time(NULL) + 5 * count;
1807
}
1808
1809
TRACE_EXIT();
1810
}
1811
1812
//! Informs the specified client (or all remote clients) that a new client is now master.
1813
void
1814
DistributionSocketLink::send_new_master(Client *client)
1815
{
1816
TRACE_ENTER("DistributionSocketLink::send_new_master");
1817
1818
PacketBuffer packet;
1819
1820
packet.create();
1821
init_packet(packet, PACKET_NEW_MASTER);
1822
1823
string id;
1824
1825
if (master_client == NULL)
1826
{
1827
// I've become master
1828
id = get_my_id();
1829
}
1830
else if (master_client->id != NULL)
1831
{
1832
// Another remote client becomes master
1833
id = master_client->id;
1834
}
1835
1836
packet.pack_string(id);
1837
packet.pack_ushort(0);
1838
1839
if (client != NULL)
1840
{
1841
send_packet(client, packet);
1842
}
1843
else
1844
{
1845
send_packet_broadcast(packet);
1846
}
1847
1848
TRACE_EXIT();
1849
1850
}
1851
1852
1853
//! Handles a new master event.
1854
void
1855
DistributionSocketLink::handle_new_master(PacketBuffer &packet, Client *client)
1856
{
1857
TRACE_ENTER("DistributionSocketLink::handle_new_master");
1858
1859
for (list<Client *>::iterator i = clients.begin(); i != clients.end(); i++)
1860
{
1861
(*i)->reject_count = 0;
1862
}
1863
1864
gchar *id = packet.unpack_string();
1865
/* gint count = */ packet.unpack_ushort();
1866
1867
dist_manager->log(_("Client %s is now the new master."),
1868
id == NULL ? "Unknown" : id);
1869
1870
if (client->id != NULL)
1871
{
1872
TRACE_MSG("new master from " << client->id << " -> " << id);
1873
}
1874
1875
set_master_by_id(id);
1876
1877
g_free(id);
1878
1879
TRACE_EXIT();
1880
}
1881
1882
1883
// Distributes the current client message.
1884
void
1885
DistributionSocketLink::send_client_message(DistributionClientMessageType type)
1886
{
1887
TRACE_ENTER("DistributionSocketLink:send_client_message");
1888
1889
PacketBuffer packet;
1890
packet.create();
1891
init_packet(packet, PACKET_CLIENTMSG);
1892
1893
string id = get_master();
1894
packet.pack_string(id);
1895
1896
packet.pack_ushort(client_message_map.size());
1897
1898
ClientMessageMap::iterator i = client_message_map.begin();
1899
while (i != client_message_map.end())
1900
{
1901
DistributionClientMessageID id = i->first;
1902
ClientMessageListener &sl = i->second;
1903
1904
IDistributionClientMessage *itf = sl.listener;
1905
1906
int pos = 0;
1907
packet.pack_ushort(id);
1908
packet.reserve_size(pos);
1909
1910
if ((sl.type & type) != 0)
1911
{
1912
TRACE_MSG("request " << id << " " << type);
1913
itf->request_client_message(id, packet);
1914
}
1915
1916
packet.update_size(pos);
1917
1918
i++;
1919
}
1920
1921
send_packet_broadcast(packet);
1922
TRACE_EXIT();
1923
}
1924
1925
1926
//! Handles client message from a remote client.
1927
void
1928
DistributionSocketLink::handle_client_message(PacketBuffer &packet, Client *client)
1929
{
1930
TRACE_ENTER("DistributionSocketLink:handle_client_message");
1931
(void) client;
1932
1933
bool will_i_become_master = false;
1934
1935
// dist_manager->log(_("Reveived client message from client %s:%d."), client->hostname, client->port);
1936
1937
gchar *id = packet.unpack_string();
1938
1939
if (id != NULL)
1940
{
1941
//TRACE_MSG("id = " << id);
1942
1943
will_i_become_master = client_is_me(id);
1944
g_free(id);
1945
}
1946
1947
gint size = packet.unpack_ushort();
1948
int pos;
1949
1950
TRACE_MSG("size = " << size);
1951
for (int i = 0; i < size; i++)
1952
{
1953
DistributionClientMessageID id = (DistributionClientMessageID) packet.unpack_ushort();
1954
gint datalen = packet.read_size(pos);
1955
1956
TRACE_MSG("len = " << datalen << " " << id);
1957
1958
if (datalen != 0)
1959
{
1960
// Narrow the buffer to the client message data.
1961
packet.narrow(-1, datalen);
1962
1963
ClientMessageMap::iterator it = client_message_map.find(id);
1964
if (it != client_message_map.end())
1965
{
1966
client_message_map[id].listener->client_message(id, will_i_become_master, client->id, packet);
1967
}
1968
1969
packet.narrow(0, -1);
1970
}
1971
1972
packet.skip_size(pos);
1973
}
1974
1975
TRACE_EXIT();
1976
}
1977
1978
1979
bool
1980
DistributionSocketLink::start_async_server()
1981
{
1982
TRACE_ENTER("DistributionSocketLink::start_async_server");
1983
bool ret = false;
1984
1985
try
1986
{
1987
/* Create the server */
1988
server_socket = socket_driver->create_server();
1989
1990
if (server_socket != NULL)
1991
{
1992
server_socket->set_listener(this);
1993
server_socket->listen(server_port);
1994
dist_manager->log(_("Network operation started."));
1995
ret = true;
1996
}
1997
}
1998
catch(SocketException e)
1999
{
2000
}
2001
2002
TRACE_RETURN(ret);
2003
return ret;
2004
}
2005
2006
2007
void
2008
DistributionSocketLink::socket_accepted(ISocketServer *scon, ISocket *ccon)
2009
{
2010
(void) scon;
2011
2012
TRACE_ENTER("DistributionSocketLink::socket_accepted");
2013
if (ccon != NULL)
2014
{
2015
dist_manager->log(_("Accepted new client."));
2016
2017
Client *client = new Client;
2018
client->type = CLIENTTYPE_DIRECT;
2019
client->peer = NULL;
2020
client->packet.create();
2021
2022
TRACE_RETURN(client->packet.bytes_available());
2023
client->socket = ccon;
2024
client->hostname = NULL;
2025
client->id = NULL;
2026
client->port = 0;
2027
client->reconnect_count = 0;
2028
client->reconnect_time = 0;
2029
2030
ccon->set_data(client);
2031
ccon->set_listener(this);
2032
clients.push_back(client);
2033
}
2034
TRACE_EXIT();
2035
}
2036
2037
2038
void
2039
DistributionSocketLink::socket_io(ISocket *con, void *data)
2040
{
2041
TRACE_ENTER("DistributionSocketLink::socket_io");
2042
bool ret = true;
2043
2044
Client *client = (Client *)data;
2045
g_assert(client != NULL);
2046
2047
TRACE_MSG("1");
2048
2049
if (!is_client_valid(client) && client->type == CLIENTTYPE_DIRECT)
2050
{
2051
TRACE_RETURN("Invalid client");
2052
return;
2053
}
2054
2055
int bytes_read = 0;
2056
int bytes_to_read = 4;
2057
2058
TRACE_MSG("2 " << client->packet.bytes_available() );
2059
2060
if (client->packet.bytes_available() >= 4)
2061
{
2062
TRACE_MSG("3");
2063
2064
bytes_to_read = client->packet.peek_ushort(0) - 4;
2065
2066
TRACE_MSG("4 " << bytes_to_read);
2067
2068
if (bytes_to_read + 4 > client->packet.get_buffer_size())
2069
{
2070
TRACE_MSG("5 " << bytes_to_read << " " << client->packet.get_buffer_size());
2071
// FIXME: the 1024 is lame...
2072
client->packet.resize(bytes_to_read + 4 + 1024);
2073
}
2074
}
2075
2076
TRACE_MSG("5");
2077
bool ok = true;
2078
try
2079
{
2080
con->read(client->packet.get_write_ptr(), bytes_to_read, bytes_read);
2081
}
2082
catch (SocketException)
2083
{
2084
ok = false;
2085
}
2086
2087
if (!ok)
2088
{
2089
dist_manager->log(_("Client %s read error, closing."),
2090
client->id == NULL ? "Unknown" : client->id);
2091
ret = false;
2092
}
2093
else if (bytes_read == 0)
2094
{
2095
dist_manager->log(_("Client %s closed connection."),
2096
client->id == NULL ? "Unknown" : client->id);
2097
ret = false;
2098
}
2099
else
2100
{
2101
g_assert(bytes_read > 0);
2102
client->packet.write_ptr += bytes_read;
2103
2104
if (client->packet.peek_ushort(0) == client->packet.bytes_written())
2105
{
2106
process_client_packet(client);
2107
}
2108
}
2109
2110
if (!ret)
2111
{
2112
close_client(client, client->outbound);
2113
}
2114
2115
TRACE_EXIT();
2116
return;
2117
}
2118
2119
2120
void
2121
DistributionSocketLink::socket_connected(ISocket *con, void *data)
2122
{
2123
TRACE_ENTER("DistributionSocketLink::socket_connected");
2124
2125
Client *client = (Client *)data;
2126
2127
g_assert(client != NULL);
2128
g_assert(con != NULL);
2129
2130
if (!is_client_valid(client) && client->type == CLIENTTYPE_DIRECT)
2131
{
2132
TRACE_RETURN("Invalid client");
2133
return;
2134
}
2135
2136
dist_manager->log(_("Client %s connected."),
2137
client->id != NULL ? client->id : "Unknown");
2138
2139
client->reconnect_count = 0;
2140
client->reconnect_time = 0;
2141
client->outbound = true;
2142
client->socket = con;
2143
2144
send_hello(client);
2145
2146
TRACE_EXIT();
2147
}
2148
2149
2150
void
2151
DistributionSocketLink::socket_closed(ISocket *con, void *data)
2152
{
2153
TRACE_ENTER("DistributionSocketLink::socket_closed");
2154
(void) con;
2155
2156
Client *client = (Client *)data;
2157
assert(client != NULL);
2158
2159
if (!is_client_valid(client) && client->type == CLIENTTYPE_DIRECT)
2160
{
2161
TRACE_RETURN("Invalid client");
2162
return;
2163
}
2164
2165
// Socket error. Disable client.
2166
if (client->socket != NULL)
2167
{
2168
dist_manager->log(_("Client %s closed connection."),
2169
client->id != NULL ? client->id : "Unknown");
2170
close_client(client, client->outbound);
2171
}
2172
else
2173
{
2174
dist_manager->log(_("Could not connect to client %s."),
2175
client->id != NULL ? client->id : "Unknown");
2176
remove_client(client);
2177
}
2178
2179
TRACE_EXIT();
2180
}
2181
2182
2183
//! Read the configuration from the configurator.
2184
void
2185
DistributionSocketLink::read_configuration()
2186
{
2187
int old_port = server_port;
2188
2189
const char *port = getenv("WORKRAVE_PORT");
2190
if (port != NULL)
2191
{
2192
server_port = atoi(port);
2193
}
2194
else
2195
{
2196
server_port = dist_manager->get_port();
2197
}
2198
2199
if (old_port != server_port && server_enabled)
2200
{
2201
set_server_enabled(false);
2202
set_server_enabled(true);
2203
}
2204
2205
reconnect_interval = dist_manager->get_reconnect_interval();
2206
reconnect_attempts = dist_manager->get_reconnect_attempts();
2207
2208
string str;
2209
str = dist_manager->get_username();
2210
username = str != "" ? g_strdup(str.c_str()) : NULL;
2211
2212
str = dist_manager->get_password();
2213
password = str != "" ? g_strdup(str.c_str()) : NULL;
2214
}
2215
2216
2217
//! Notification from the configurator that the configuration has changed.
2218
void
2219
DistributionSocketLink::config_changed_notify(const string &key)
2220
{
2221
(void) key;
2222
read_configuration();
2223
}
2224
2225
#endif
Loggerhead is a web-based interface for Breezy
Version: 2.0.1