2 {Upload Manager for brodnetd}
5 USES Chat
,TC
,opcode
,ServerLoop
,MemStream
,NetAddr
;
19 isOpen
,Active
:boolean;
21 oinfo
:tStoreObjectInfo
;
22 procedure Init(ag
:tAggr_ptr
; var nchat
:tChat
; msg
: tSMsg
);
23 procedure OnMsg(msg
:tSMsg
; data
:boolean);
24 procedure IdleTimeout
;
25 procedure ChatTimeout(willwait
:LongWord
);
27 procedure DoGET(const fid
:tfid
; base
,limit
:LongWord
);
28 procedure DoSEG(base
,limit
:LongWord
);
41 procedure Init(const source
:tNetAddr
);
50 GET(filehash:20; baseHi:word2; base:word4; limit:word4);
51 SEG(baseHi:word2; base:word4; limit:word4);
53 INFO(sizeHi:word2; size:word4; final:byte; seglen:word4);
58 procedure tPrv
.DoGET(const fid
:tfid
; base
,limit
:LongWord
);
59 var err
:tmemorystream
;
61 if isOpen
then oinfo
.Close
;
62 if Active
then Stop
; //opt
65 {if not oinfo.final then begin
69 if oinfo
.rc
>0 then begin
70 ch
^.StreamInit(err
,3);
71 err
.WriteByte(upFAIL
);
72 if oinfo
.rc
=1 then err
.WriteByte(upErrNotFound
)
73 else begin err
.WriteByte(upErrIO
); err
.WriteByte(oinfo
.rc
) end;
81 procedure tPrv
.DoSEG(base
,limit
:LongWord
);
82 var err
:tmemorystream
;
85 ch
^.StreamInit(err
,12);
87 if oinfo
.rc
>0 then begin
88 err
.WriteByte(upFAIL
);
89 err
.WriteByte(upErrIO
);
90 err
.WriteByte(oinfo
.rc
);
94 err
.WriteByte(upINFO
);
96 err
.WriteWord(oinfo
.length
,4);
98 if oinfo
.seglen
<seglen
then seglen
:=oinfo
.seglen
;
100 then err
.WriteByte(1)
101 else err
.WriteByte(0);
102 err
.WriteWord(seglen
,4);
104 if not Active
then Start
;
106 ch
^.StreamInit(err
,2);
107 err
.WriteByte(upFAIL
);
108 err
.WriteByte(upErrSegNoGet
);
113 procedure tPrv
.Start
;
118 UnShedule(@IdleTimeout
);
119 writeln('upmgr: Startig transfer');
120 if not assigned(aggr
^.prv
) then begin
126 next
:=aggr
^.prv
^.next
;
138 if prev
<>@self
then begin
142 if aggr
^.prv
=@self
then aggr
^.prv
:=next
;
144 Shedule(20000,@IdleTimeout
);
145 writeln('upmgr: Stop');
152 var buf
:array [1..2048] of byte;
154 // writeln('upmgr: CONT! ',chan);
155 Assert(Active
and isOpen
);
156 sz
:=aggr
^.tcs
.MaxSize(sizeof(buf
))-1;
157 if sz
>SegLen
then sz
:=SegLen
;
158 //s.Init(GetMem(sz),0,sz);
159 Assert((sz
+1)<=sizeof(buf
));
160 s
.Init(@buf
,0,sizeof(buf
)); aggr
^.tcs
.WriteHeaders(s
);
162 Assert(sz
<=s
.WrBufLen
);
163 oinfo
.ReadAhead(sz
,s
.WrBuf
); //todo
165 Assert(oinfo
.rc
=0); //todo
168 //FreeMem(s.base,s.size);
171 if SegLen
=0 then begin
177 if (wcur
=0) then begin
183 procedure tPrv
.DoClose
;
186 if isOpen
then oinfo
.Close
;
188 UnShedule(@IdleTimeout
);
192 FreeMem(@self
,sizeof(self
));
195 procedure tPrv
.OnMsg(msg
:tSMsg
; data
:boolean);
200 var err
:tmemorystream
;
203 if not data
then exit
; //todo
204 Assert(not(aggr
^.rekt
and active
));
205 if aggr
^.rekt
then exit
;
206 if msg
.stream
.RdBufLen
<1 then goto malformed
;
207 op
:=msg
.stream
.ReadByte
;
208 writeln('upmgr: opcode=',op
,' sz=',msg
.stream
.RdBufLen
);
212 if msg
.stream
.RdBufLen
<>30 then goto malformed
;
213 msg
.stream
.Read(hash
,20);
214 if msg
.stream
.ReadWord(2)>0 then goto malformed
;
215 base
:=msg
.stream
.ReadWord(4);
216 limit
:=msg
.stream
.ReadWord(4);
217 DoGet(hash
,base
,limit
);
220 if msg
.stream
.RdBufLen
<10 then goto malformed
;
221 if msg
.stream
.ReadWord(2)>0 then goto malformed
;
222 base
:=msg
.stream
.ReadWord(4);
223 limit
:=msg
.stream
.ReadWord(4);
229 ch
^.StreamInit(err
,2);
230 err
.WriteByte(upFAIL
);
231 err
.WriteByte(upErrMalformed
);
233 writeln('upmgr: malformed request stage=1');
236 procedure tPrv
.ChatTimeout(willwait
:LongWord
);
237 var wasactive
:boolean;
239 if WillWait
<8000 then exit
;
240 writeln('upmgr: Chat timeout');
243 if isOpen
then oinfo
.Close
;
247 if wasactive
then IdleTimeout
{else it is sheduled};
249 procedure tPrv
.IdleTimeout
;
250 var err
:tMemoryStream
;
252 if assigned(ch
) then begin {chat is still not rekt}
253 if not active
then writeln('upmgr: Idle timeout');
254 ch
^.StreamInit(err
,1);
255 err
.WriteByte(upClose
);
260 {it is idle timeout, but may be called from aggr it tc tiomes out}
262 if isOpen
then oinfo
.Close
; {may still be open}
263 UnShedule(@IdleTimeout
);
265 FreeMem(@self
,sizeof(self
));
268 procedure tPrv
.Init(ag
:tAggr_ptr
; var nchat
:tChat
; msg
: tSMsg
);
271 ch
^.Callback
:=@OnMsg
;
272 ch
^.TMHook
:=@ChatTimeout
;
276 chan
:=msg
.stream
.readbyte
;
277 writeln('upmgr: prv init chan=',chan
);
280 isOpen
:=false; Active
:=false;
282 Shedule(5000,@IdleTimeout
);
286 procedure tAggr
.Init(const source
:tNetAddr
);
288 writeln('upmgr: init');
292 if assigned(Peers
) then Peers
^.prev
:=@self
;
297 tcs
.OnTimeout
:=@TCTimeout
;
302 procedure tAggr
.TCTimeout
;
305 writeln('upmgr: TCTimeout');
306 while assigned(prv
) do begin
315 procedure tAggr
.Cont
;
317 if not assigned(prv
) then exit
;
320 procedure tAggr
.UnRef
;
324 writeln('upmgr: aggr unrefd');
327 FreeMem(@self
,sizeof(self
));
330 procedure tAggr
.Done
;
333 writeln('upmgr: aggr close');
336 if assigned(prev
) then prev
^.next
:=next
else Peers
:=next
;
337 if assigned(next
) then next
^.prev
:=prev
;
340 function FindAggr({const} addr
:tNetAddr
): tAggr_ptr
;
343 while assigned(result
) do begin
344 if assigned(result
^.next
) then assert(result
^.next
^.prev
=result
);
345 if result
^.tcs
.remote
=addr
then exit
;
346 result
:=result
^.next
;
350 procedure ChatHandler(var nchat
:tChat
; msg
:tSMsg
);
356 writeln('upmgr: ChatHandler');
357 msg
.stream
.skip({the initcode}1);
358 if msg
.stream
.RdBufLen
<2 then begin
359 writeln('upmgr: malformed init');
360 nchat
.StreamInit(s
,16);
362 s
.writebyte(upErrMalformed
);
365 writeln('upmgr: malformed request stage=0');
368 ag
:=FindAggr(msg
.source
^);
369 if assigned(ag
) then begin
371 if ag
^.Cnt
>=cMax
then begin
372 nchat
.StreamInit(s
,16);
374 s
.WriteByte(upErrHiChan
);
376 s
.WriteByte(ag
^.Cnt
);
382 ag
^.init(msg
.source
^);
385 pr
^.Init(ag
,nchat
,msg
);
389 SetChatHandler(opcode
.upFileServer
,@ChatHandler
);