Discussion :

[Arnaud F.]

Bonjour,

je lis des données sur un flux réseau (issu de socket) de manière asynchrone sauf que le traitement continue avant que tous le flux soit lu.

Le code de lecture :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
 
        private TcpClient _client;
        private NetworkStream _clientStream;
 
        private MemoryStream _responseBytes;
        private readonly UTF8Encoding _UTF8Encoder = new UTF8Encoding();
        private const int BUFFER_SIZE = 8096;
 
        private XElement Receive()
        {
            byte[] buffer = new byte[BUFFER_SIZE];
 
            XElement xmlResult;
            Encoding serverEncoding = this.Task.Server.Encoding;
 
            // Reading result
 
            while (true)
            {
                _responseBytes = new MemoryStream();
 
                try
                {
                    IAsyncResult e = _clientStream.BeginRead(buffer,
                        0,                                              // Begin
                        BUFFER_SIZE,                                    // Length
                        new AsyncCallback(OnBeginRead),                 // Callback used
                        new SocketAsyncState(_clientStream, buffer)); // Passing buffer to callback
 
                    e.AsyncWaitHandle.WaitOne();    // Wait until data are in pipe
 
                    if (((SocketAsyncState)e.AsyncState).HasError)
                    {
                        throw new ObjectDisposedException();
                    }
 
                    // Try to convert to a XElement, if fail, redo all process.
                    _responseBytes.Position = 0;
 
                    try
                    {
                        xmlResult = XElement.Parse(
                            _UTF8Encoder.GetString(_responseBytes.GetBuffer(), 0, (int)_responseBytes.Length),
                            LoadOptions.PreserveWhitespace);
                    }
                    catch
                    {
                        xmlResult = XElement.Parse(
                           _UTF8Encoder.GetString(_responseBytes.GetBuffer(), 0, (int)_responseBytes.Length),
                           LoadOptions.PreserveWhitespace);
                    }
 
                    // Result 100% retrieved : quit loop
                    break;
                }
                catch (Exception ex)
                {
                    if (ex is ObjectDisposedException
                        || ex is XmlException)
                    {
                        while (!IsConnected) { Wait(); }   // Wait that the network comes back
                        SendSyn();                         // Relaunch process
                    }
                }
            }
 
            // Result 100% retrieved : send ACK to FmSocket
            SendAck();
 
            return xmlResult;
        }
 
        /// <summary>
        /// Begins an asynchronous read from the NetworkStream.
        /// </summary>
        /// <param name="ar">An IAsyncResult that represents the asynchronous call.</param>
        private void OnBeginRead(IAsyncResult ar)
        {
            SocketAsyncState state = ar.AsyncState as SocketAsyncState;
 
            byte[] bufferReaded = state.Data;
 
            state.Data = new byte[BUFFER_SIZE];
 
            int numberOfBytesReaded;
            Encoding serverEncoding = this.Task.Server.Encoding;
 
            try
            {
                numberOfBytesReaded = state.Stream.EndRead(ar);
            }
            catch(Exception)
            {
                ((SocketAsyncState)ar.AsyncState).HasError = true;
                // Quit
                return;
            }
 
            // While data are available, read next buffer (recursive call to this method)
            if (state.Stream.DataAvailable && state.Stream.CanRead)
            {
                state.Stream.BeginRead(state.Data,
                    0,
                    BUFFER_SIZE,
                    new AsyncCallback(OnBeginRead),
                    state);
            }
 
            // Default C# strings are in UTF-8, so convert stream only if needed
            if (serverEncoding.CodePage != _UTF8Encoder.CodePage)
            {
                byte[] buffer = Encoding.Convert(serverEncoding,
                    _UTF8Encoder,
                    bufferReaded.TakeWhile((b) => b != '\0').ToArray());
 
                _responseBytes.Write(buffer, 0, buffer.Length);
 
#if DEBUG
                System.Diagnostics.Debug.Write(_UTF8Encoder.GetString(buffer, 0, buffer.Length));
#endif
 
            }
            else
            {
                _responseBytes.Write(bufferReaded, 0, numberOfBytesReaded);
 
#if DEBUG
                System.Diagnostics.Debug.Write(_UTF8Encoder.GetString(bufferReaded, 0, numberOfBytesReaded));
#endif
 
            }
        }

La console :

Code XML :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
 
<?xml version="1.0" encoding="Windows-1252" standalone="yes" ?>
<return value="0">
	<resultset>
		<meta>
			<column type="char(30)"></column>
		</meta>
		<datarow>
			<datacol>
				<![CDATA[master]]>
			</datacol>
		</datarow>
	</resultset>
</return>

Le thread 0x276c531a s'est arrêté avec le code 0 (0x0).
'TestFmLibrary.exe' (Managé)*: 'System.SR.resources.dll' chargé
Une exception de première chance de type 'System.Xml.XmlException' s'est produite dans System.Xml.dll
[Fin FmSocket : 4000ms]

En fait, le WaitOne() n'attend pas la fin de la lecture car lors du parsing du résultat en un XElement, le code du try plante tandis que le code du catch fonctionne (le flux est le même ...), j'en déduis que dans le try, le flux mémoire n'est pas encore totalement reçu tandis que dans le catch, il l'est...

Pour les plus gros flux, c'est encore plus flagrant, le code continue sans attendre la fin de réception des données.

Qu'est-ce qui cloche dans mon code?