Skip to content

Commit b4e9b4e

Browse files
author
Aleksandr Sokolovskii
committed
use Isolation for sending and receiving chat messages
1 parent bd95633 commit b4e9b4e

4 files changed

+187
-103
lines changed

‎flutter-grpc-tutorial.code-workspace

+1
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
],
1616
"settings": {
1717
"cSpell.words": [
18+
"Friendlychat",
1819
"Stateful",
1920
"amsokol",
2021
"grpc",

‎flutter_client/lib/chat_message_outgoing.dart

+1-1
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ const String _name = "Me";
88
/// Outgoing message statuses
99
/// UNKNOWN - message just created and is not sent yet
1010
/// SENT - message is sent to the server successfully
11-
enum MessageOutgoingStatus { UNKNOWN, SENT }
11+
enum MessageOutgoingStatus { UNKNOWN, SENT, FAILED }
1212

1313
/// MessageOutgoing is class defining message data (id and text) and status
1414
class MessageOutgoing extends Message {

‎flutter_client/lib/chat_screen.dart

+1-1
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@ class ChatScreenState extends State<ChatScreen> with TickerProviderStateMixin {
5151
onSentError: onSentError,
5252
onReceivedSuccess: onReceivedSuccess,
5353
onReceivedError: onReceivedError);
54-
_service.startListening();
54+
_service.start();
5555
}
5656

5757
@override

‎flutter_client/lib/chat_service.dart

+184-101
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
1+
import 'dart:isolate';
2+
import 'dart:io';
13
import 'package:grpc/grpc.dart';
24

35
import 'api/v1/chat.pbgrpc.dart' as grpc;
@@ -7,19 +9,25 @@ import 'chat_message.dart';
79
import 'chat_message_outgoing.dart';
810

911
/// CHANGE TO IP ADDRESS OF YOUR SERVER IF IT IS NECESSARY
10-
const serverIP = "127.0.0.1";
12+
const serverIP = "10.0.2.2";
1113
const serverPort = 3000;
1214

1315
/// ChatService client implementation
1416
class ChatService {
15-
/// Flag is indicating that client is shutting down
16-
bool _isShutdown = false;
17+
// _isolateSending is isolate to send chat messages
18+
Isolate _isolateSending;
1719

18-
/// gRPC client channel to send messages to the server
19-
ClientChannel _clientSend;
20+
// Port to send message
21+
SendPort _portSending;
2022

21-
/// gRPC client channel to receive messages from the server
22-
ClientChannel _clientReceive;
23+
// Port to get status of message sending
24+
ReceivePort _portSendStatus;
25+
26+
// _isolateReceiving is isolate to receive chat messages
27+
Isolate _isolateReceiving;
28+
29+
// Port to receive messages
30+
ReceivePort _portReceiving;
2331

2432
/// Event is raised when message has been sent to the server successfully
2533
final void Function(MessageOutgoing message) onSentSuccess;
@@ -38,117 +46,192 @@ class ChatService {
3846
{this.onSentSuccess,
3947
this.onSentError,
4048
this.onReceivedSuccess,
41-
this.onReceivedError});
42-
43-
// Shutdown client
44-
Future<void> shutdown() async {
45-
_isShutdown = true;
46-
_shutdownSend();
47-
_shutdownReceive();
49+
this.onReceivedError})
50+
: _portSendStatus = ReceivePort(),
51+
_portReceiving = ReceivePort();
52+
53+
// Start threads to send and receive messages
54+
void start() {
55+
_startSending();
56+
_startReceiving();
4857
}
4958

50-
// Shutdown client (send channel)
51-
void _shutdownSend() {
52-
if (_clientSend != null) {
53-
_clientSend.shutdown();
54-
_clientSend = null;
59+
/// Start thread to send messages
60+
void _startSending() async {
61+
// start thread to send messages
62+
_isolateSending =
63+
await Isolate.spawn(_sendingIsolate, _portSendStatus.sendPort);
64+
65+
// listen send status
66+
await for (var msg in _portSendStatus) {
67+
if (msg is SendPort) {
68+
_portSending = msg;
69+
} else {
70+
var message = msg[0] as MessageOutgoing;
71+
var statusDetails = msg[1] as String;
72+
switch (message.status) {
73+
case MessageOutgoingStatus.SENT:
74+
// call for success handler
75+
if (onSentSuccess != null) {
76+
onSentSuccess(message);
77+
}
78+
break;
79+
case MessageOutgoingStatus.FAILED:
80+
// call for error handler
81+
if (onSentError != null) {
82+
onSentError(message, statusDetails);
83+
}
84+
break;
85+
default:
86+
// call for error handler
87+
if (onSentError != null) {
88+
onSentError(message, "unexpected message status");
89+
}
90+
break;
91+
}
92+
}
5593
}
5694
}
5795

58-
// Shutdown client (receive channel)
59-
void _shutdownReceive() {
60-
if (_clientReceive != null) {
61-
_clientReceive.shutdown();
62-
_clientReceive = null;
63-
}
64-
}
96+
/// Thread to send messages
97+
static void _sendingIsolate(SendPort portSendStatus) async {
98+
// Port to get messages to send
99+
ReceivePort portSendMessages = ReceivePort();
100+
101+
// send port to send messages to the caller
102+
portSendStatus.send(portSendMessages.sendPort);
103+
104+
ClientChannel client;
105+
106+
// waiting messages to send
107+
await for (MessageOutgoing message in portSendMessages) {
108+
var sent = false;
109+
do {
110+
if (client == null) {
111+
// create new client
112+
client = ClientChannel(
113+
serverIP, // Your IP here or localhost
114+
port: serverPort,
115+
options: ChannelOptions(
116+
//TODO: Change to secure with server certificates
117+
credentials: ChannelCredentials.insecure(),
118+
idleTimeout: Duration(seconds: 1),
119+
),
120+
);
121+
}
65122

66-
/// Send message to the server
67-
void send(MessageOutgoing message) {
68-
if (_clientSend == null) {
69-
// create new client
70-
_clientSend = ClientChannel(
71-
serverIP, // Your IP here or localhost
72-
port: serverPort,
73-
options: ChannelOptions(
74-
//TODO: Change to secure with server certificates
75-
credentials: ChannelCredentials.insecure(),
76-
idleTimeout: Duration(seconds: 10),
77-
),
78-
);
79-
}
123+
MessageOutgoingStatus statusCode;
124+
String statusDetails;
125+
126+
try {
127+
// try to send
128+
var request = StringValue.create();
129+
request.value = message.text;
130+
await grpc.ChatServiceClient(client).send(request);
131+
// sent successfully
132+
statusCode = MessageOutgoingStatus.SENT;
133+
sent = true;
134+
} catch (e) {
135+
// sent failed
136+
statusCode = MessageOutgoingStatus.FAILED;
137+
statusDetails = e.toString();
138+
// reset client
139+
client.shutdown();
140+
client = null;
141+
} finally {
142+
var msg = MessageOutgoing(
143+
text: message.text, id: message.id, status: statusCode);
144+
portSendStatus.send([
145+
msg,
146+
statusDetails,
147+
]);
148+
}
80149

81-
var request = StringValue.create();
82-
request.value = message.text;
83-
84-
grpc.ChatServiceClient(_clientSend).send(request).then((_) {
85-
// call for success handler
86-
if (onSentSuccess != null) {
87-
var sentMessage = MessageOutgoing(
88-
text: message.text,
89-
id: message.id,
90-
status: MessageOutgoingStatus.SENT);
91-
onSentSuccess(sentMessage);
92-
}
93-
}).catchError((e) {
94-
if (!_isShutdown) {
95-
// invalidate current client
96-
_shutdownSend();
150+
if (!sent) {
151+
// try to send again
152+
sleep(Duration(seconds: 5));
153+
}
154+
} while (!sent);
155+
}
156+
}
97157

158+
/// Start listening messages from the server
159+
void _startReceiving() async {
160+
// start thread to receive messages
161+
_isolateReceiving =
162+
await Isolate.spawn(_receivingIsolate, _portReceiving.sendPort);
163+
164+
// listen for incoming messages
165+
await for (var msg in _portReceiving) {
166+
var message = msg[0] as Message;
167+
var error = msg[1] as String;
168+
if (error != null) {
98169
// call for error handler
99-
if (onSentError != null) {
100-
onSentError(message, e.toString());
170+
if (onReceivedError != null) {
171+
onReceivedError(error);
172+
}
173+
} else {
174+
if (onReceivedSuccess != null) {
175+
onReceivedSuccess(message);
101176
}
102-
103-
// try to send again
104-
Future.delayed(Duration(seconds: 30), () {
105-
send(message);
106-
});
107177
}
108-
});
178+
}
109179
}
110180

111-
/// Start listening messages from the server
112-
void startListening() {
113-
if (_clientReceive == null) {
114-
// create new client
115-
_clientReceive = ClientChannel(
116-
serverIP, // Your IP here or localhost
117-
port: serverPort,
118-
options: ChannelOptions(
119-
//TODO: Change to secure with server certificates
120-
credentials: ChannelCredentials.insecure(),
121-
idleTimeout: Duration(seconds: 10),
122-
),
123-
);
124-
}
181+
/// Thread to listen messages from the server
182+
static void _receivingIsolate(SendPort portReceive) async {
183+
ClientChannel client;
184+
185+
do {
186+
if (client == null) {
187+
// create new client
188+
client = ClientChannel(
189+
serverIP, // Your IP here or localhost
190+
port: serverPort,
191+
options: ChannelOptions(
192+
//TODO: Change to secure with server certificates
193+
credentials: ChannelCredentials.insecure(),
194+
idleTimeout: Duration(seconds: 1),
195+
),
196+
);
197+
}
125198

126-
var stream =
127-
grpc.ChatServiceClient(_clientReceive).subscribe(Empty.create());
199+
var stream = grpc.ChatServiceClient(client).subscribe(Empty.create());
128200

129-
stream.forEach((msg) {
130-
if (onReceivedSuccess != null) {
131-
var message = Message(msg.text);
132-
onReceivedSuccess(message);
201+
try {
202+
await for (var msg in stream) {
203+
var message = Message(msg.text);
204+
portReceive.send([message, null]);
205+
}
206+
} catch (e) {
207+
// reset client
208+
client.shutdown();
209+
client = null;
210+
// notify caller
211+
portReceive.send([null, e.toString()]);
133212
}
134-
}).then((_) {
135-
// raise exception to start listening again
136-
throw Exception("stream from the server has been closed");
137-
}).catchError((e) {
138-
if (!_isShutdown) {
139-
// invalidate current client
140-
_shutdownReceive();
213+
// try to connect again
214+
sleep(Duration(seconds: 5));
215+
} while (true);
216+
}
141217

142-
// call for error handler
143-
if (onReceivedError != null) {
144-
onReceivedError(e.toString());
145-
}
218+
// Shutdown client
219+
void shutdown() {
220+
// stop sending
221+
_isolateSending?.kill(priority: Isolate.immediate);
222+
_isolateSending = null;
223+
_portSendStatus?.close();
224+
_portSendStatus = null;
225+
226+
// stop receiving
227+
_isolateReceiving?.kill(priority: Isolate.immediate);
228+
_isolateReceiving = null;
229+
_portReceiving?.close();
230+
_portReceiving = null;
231+
}
146232

147-
// start listening again
148-
Future.delayed(Duration(seconds: 30), () {
149-
startListening();
150-
});
151-
}
152-
});
233+
/// Send message to the server
234+
void send(MessageOutgoing message) {
235+
_portSending.send(message);
153236
}
154237
}

0 commit comments

Comments
 (0)