UpMgr work + Tests and temp store.
[brdnet.git] / Chat.pas
blobf0af62d64616824bcc02c1415221ed624d955fcd
1 unit Chat;
3 Implement two-way realiable acked lock-step protocol
5 INTERFACE
6 uses NetAddr,ServerLoop,MemStream;
8 type tChat=object
9 remote:tNetAddr;
10 opcode:byte;
11 txSeq:Word;
12 rxSeq:Word;
13 rxAcked:boolean;
14 closed:boolean;
15 RTT:LongWord;{in ms}
16 callback: procedure(msg:tSMsg; data:boolean) of object; {client must maintain active chats}
17 TMhook : procedure(willwait:LongWord ) of object;
18 DisposeHook: procedure of object; {called instead of freeing self}
19 procedure Init(const iremote:tNetAddr);
20 procedure AddHeaders(var s:tMemoryStream);
21 procedure StreamInit(var s:tMemoryStream; l:word);
22 procedure Send(s:tMemoryStream);
23 {the stream can be invalidated, but the buffer must not be modified or freed}
24 procedure Ack;
25 procedure Close;
26 private
27 txPk:pointer; txLen:word; {last sent, not acked msg}
28 txTime:tDateTime;
29 procedure InitFrom(const iremote:tNetAddr; iopcode:byte);
30 procedure Done;
31 procedure Resend;
32 procedure OnReply(msg:tSMsg);
33 end;
35 type tChatHandler=procedure(var nchat:tChat; msg:tSMsg);
36 procedure SetChatHandler(initcode:byte; handler:tChatHandler);
38 { download manager create FileRequest
39 File Request open chat session to server
40 upmgr accepts chat and send reply
41 FileRequest acks, chat is then closed after TimeWait
42 upmgr starts TC transfer
43 transfer finished, upmgr send new Chat to FileRequest
44 FileRequest acks, chat is closed on both ends
45 FileRequest can open new chat if blocks are missing
47 => chat msgs must be created with New, disposed by Chat
48 => there is TimeWait, no references are to the Chat, except Sheduler, it Disposes itself.
51 { Chats are the HiMsg. Use hash table from ServerLoop, works for HiMsg too. }
53 IMPLEMENTATION
54 uses SysUtils;
55 procedure tChat.Init(const iremote:tNetAddr);
56 begin
57 remote:=iremote;
58 opcode:=128;
59 while ServerLoop.IsMsgHandled(opcode,remote) do inc(opcode);
60 InitFrom(remote,opcode);
61 end;
62 procedure tChat.InitFrom(const iremote:tNetAddr; iopcode:byte);
63 begin
64 remote:=iremote;
65 opcode:=iopcode;
66 SetMsgHandler(opcode,remote,@OnReply);
67 txSeq:=0;
68 rxSeq:=0;
69 rxAcked:=true; {to not ack pk 0}
70 closed:=false;
71 txPk:=nil;
72 txLen:=0;
73 callback:=nil;
74 TMhook:=nil;
75 DisposeHook:=nil;
76 RTT:=200; {a default for timeouts}
77 txTime:=0;
78 end;
79 {struct
80 opcode:byte
81 seq:2
82 ack_seq:2
83 data:xx
86 procedure tCHat.AddHeaders(var s:tMemoryStream);
87 begin s.skip(5) end;
88 procedure tChat.StreamInit(var s:tMemoryStream; l:word);
89 begin
90 s.Init(GetMem(l+5),0,l+5);
91 AddHeaders(s);
92 end;
94 procedure tChat.Send(s:tMemoryStream);
95 begin
96 assert(txLen=0);
97 //assert(assigned(callback));
98 Inc(txSeq);
99 s.Seek(0);
100 s.WriteByte(opcode);
101 s.WriteWord(txSeq,2);
102 if not rxAcked then begin
103 s.WriteWord(rxSeq,2);
104 rxAcked:=true;
105 end else s.WriteWord(0,2);
106 txPk:=s.base;
107 txLen:=s.Length;
108 ServerLoop.SendMessage(txPk^,txLen,remote);
109 ServerLoop.Shedule(RTT*2,@Resend);
110 txTime:=Now;
111 end;
113 procedure tChat.Ack;
114 var s:tMemoryStream;
115 begin
116 if not rxAcked then begin
117 s.Init(GetMem(5),0,5);
118 s.WriteByte(opcode);
119 s.WriteWord(0,2);
120 s.WriteWord(rxSeq,2);
121 ServerLoop.SendMessage(s.base^,s.length,remote);
122 FreeMem(s.base,s.length);
123 rxAcked:=true;
124 end;
125 end;
127 procedure tChat.Close;
128 begin
129 assert(not closed);
130 Ack;
131 closed:=true;
132 //writeln('Chat: closing');
133 if txLen=0 {no packets in flight} then begin
134 Shedule(15000{todo},@Done); {wait for something lost}
135 callback:=nil; {avoid calling}
136 tmhook:=nil;
137 end;
138 end;
140 procedure tChat.Done;
141 begin
142 {called from sheduler, Done is unsheduled, Resend is not sheduled since ack was received when Done was sheduled}
143 if txLen>0 then FreeMem(txPk,txLen);
144 SetMsgHandler(opcode,remote,nil);
145 if assigned(DisposeHook) then DisposeHook
146 else FreeMem(@self,sizeof(self));
147 //writeln('Chat: closed');
148 end;
150 procedure tChat.Resend;
151 {timeout waiting for ack}
152 begin
153 {resend and reshedule}
154 if txLen=0 then exit;
155 txTime:=0;
156 if RTT<1 then RTT:=2;
157 RTT:=RTT*2;
158 if assigned(TMhook) and (not closed) then begin
159 TMhook(RTT);
160 if closed then begin
161 Done; {if hook decided to close then abort}
162 exit;
163 end;
164 end;
165 if closed and (RTT<400) then RTT:=400;
166 if (RTT>=5000) and closed then begin
167 Done {give up}
168 end else begin
169 {finally resend the msg}
170 //writeln('Chat: retry');
171 ServerLoop.SendMessage(txPk^,txLen,remote);
172 ServerLoop.Shedule(RTT,@Resend);
173 end;
174 end;
176 procedure tChat.OnReply(msg:tSMsg);
177 var seq,aseq:Word;
178 var s:tMemoryStream;
179 begin
180 msg.stream.skip(1{opcode});
181 seq:=msg.stream.ReadWord(2);
182 aseq:=msg.stream.ReadWord(2);
183 if aseq>0 then {ack of our msg} begin
184 if (aseq=txSeq)and(txLen>0) {it is current} then begin
185 if txTime>0 then RTT:=Round((Now-txTime)*MsecsPerDay);
186 FreeMem(txPk,txLen);
187 TxLen:=0;
188 txPk:=nil;
189 if assigned(callback) then callback(msg,false);
190 ServerLoop.UnShedule(@Resend);
191 end else {write(' old-ack')it is ack of old data, do nothing};
192 end;
193 if seq>0 then {some data} begin
194 if seq<=rxSeq then {remote didnt get our ack} begin
195 s.Init(GetMem(5),0,5);
196 s.WriteByte(opcode);
197 s.WriteWord(0,2);
198 s.WriteWord(rxSeq,2);
199 ServerLoop.SendMessage(s.base^,s.length,remote);
200 FreeMem(s.base,s.length);
201 if seq=rxSeq then rxacked:=true;
202 end else begin
203 {some useful data!}
204 rxSeq:=seq;
205 rxAcked:=false;
206 if assigned(callback) then callback(msg,true);
207 end;
208 end;
209 end;
211 var ChatHandlers: array [1..32] of tChatHandler;
213 procedure SetChatHandler(initcode:byte; handler:tChatHandler);
214 begin
215 assert(ChatHandlers[initcode]=nil);
216 ChatHandlers[initcode]:=handler;
217 end;
219 procedure OnHiMsg(msg:tSMsg);
220 {new chat was received!}
221 var opcode:byte;
222 var seq,aseq:word;
223 var hnd:tChatHandler;
224 var nchat:^tChat;
225 var ix:byte;
226 begin
227 opcode:=msg.stream.ReadByte;
228 assert(not IsMsgHandled(opcode,msg.source^));
229 seq:=msg.stream.ReadWord(2);
230 aseq:=msg.stream.ReadWord(2);
231 if (seq<>1)or(aseq>0) then exit; {invalid initial state}
232 ix:=msg.stream.ReadByte;
233 if (ix<1)or(ix>high(ChatHandlers)) then exit;
234 hnd:=ChatHandlers[ix];
235 if not assigned(hnd) then raise eXception.Create('No handler for initcode '+IntToStr(ix));
236 msg.stream.seek(msg.stream.position-1);{unskip the initcode}
237 nchat:=GetMem(sizeof(tChat));
238 nchat^.InitFrom(msg.Source^,opcode);
239 nchat^.rxacked:=false;
240 nchat^.rxSeq:=1;
241 hnd(nchat^,msg);
242 end;
244 BEGIN
245 FillChar(ChatHandlers,sizeof(chathandlers),0);
246 ServerLoop.SetHiMsgHandler(@OnHiMsg);
247 END.