
DIRECTORY
CommTimer,
Convert, -- debug only
CStrings,
FinalizeOps,
IO,
NetworkStream,
Process,
RefText, -- debug only
Rope,
UXStrings -- debug only
;

NetworkStreamImpl: CEDAR MONITOR
IMPORTS CommTimer, Convert, FinalizeOps, IO, NetworkStream, Process, RefText, UXStrings
EXPORTS NetworkStream
~ {
ROPE: TYPE ~ Rope.ROPE;
STREAM: TYPE ~ IO.STREAM;
sendSoonGrainSizeMsec: INT Ź 50;
sendSoonExpectedWaitMsec: INT Ź 2000;
Error: PUBLIC ERROR [which: REF, codes: LIST OF ATOM, msg: ROPE] ~ CODE;

Timeout: PUBLIC SIGNAL [which: REF, codes: LIST OF ATOM, msg: ROPE] Ź CODE;

Self: PROC [] RETURNS [PROCESS] ~ TRUSTED INLINE {
RETURN [ LOOPHOLE[ Process.GetCurrent[] ] ] };

ErrorDetails: TYPE ~ REF ErrorDetailsRecord;
ErrorDetailsRecord: TYPE ~ RECORD [
next: ErrorDetails,
process: PROCESS,
codes: LIST OF ATOM,
msg: ROPE
];

nilStreamData: NetworkStream.NetworkStreamData Ź NEW[
NetworkStream.NetworkStreamDataRecord Ź [
registration~NIL,
procs~NIL
]
];

AddErrorDetails: ENTRY PROC [d: NetworkStream.NetworkStreamData, codes: LIST OF ATOM, msg: ROPE] ~ {
IF d = NIL THEN RETURN;
d.pendingErrors Ź NEW[ErrorDetailsRecord Ź [
next~NARROW[d.pendingErrors],
process~Self[],
codes~codes,
msg~msg
]];
};

DeleteErrorDetails: ENTRY PROC [d: NetworkStream.NetworkStreamData] ~ {
p, prev: ErrorDetails Ź NIL;
self: PROCESS;
IF d = NIL THEN RETURN;
self Ź Self[];
p Ź NARROW[d.pendingErrors];
DO
IF p = NIL THEN RETURN;
IF p.process = self THEN EXIT;
prev Ź p; p Ź p.next;
ENDLOOP;
IF prev = NIL THEN d.pendingErrors Ź p.next ELSE prev.next Ź p.next;
p.next Ź NIL;  -- helps GC
};

LookupErrorDetails: ENTRY PROC [d: NetworkStream.NetworkStreamData] RETURNS [codes: LIST OF ATOM, msg: ROPE] ~ {
self: PROCESS;
p: ErrorDetails;
IF Process.GetCurrent[] # Process.GetCurrent[] THEN ERROR;
IF d = NIL THEN RETURN [codes~NIL, msg~NIL];
self Ź Self[];
p Ź NARROW[d.pendingErrors];
DO
IF p = NIL THEN RETURN [codes~NIL, msg~NIL];
IF p.process = self THEN RETURN [codes~p.codes, msg~p.msg];
p Ź p.next;
ENDLOOP;
};

GetIOErrorDetails: PUBLIC PROC [which: IO.STREAM]
RETURNS [codes: LIST OF ATOM, msg: ROPE] ~ {
d: NetworkStream.NetworkStreamData;
d Ź IF which # NIL THEN NARROW[which.streamData] ELSE nilStreamData;
[codes, msg] Ź LookupErrorDetails[d];
};

RaiseIOError: PUBLIC PROC [ec: IO.ErrorCode, stream: STREAM, codes: LIST OF ATOM, msg: ROPE] ~ {
d: NetworkStream.NetworkStreamData;
d Ź IF stream # NIL THEN NARROW[stream.streamData] ELSE nilStreamData;
{
ENABLE UNWIND => { DeleteErrorDetails[d] };
AddErrorDetails[d, codes, msg];
SELECT ec FROM
Failure => {
IF codes = NIL THEN ERROR IO.Error[ec, stream];
SELECT codes.first FROM
$timeout => {
IF NOT d.signalTimeout THEN ERROR IO.Error[ec, stream];
SIGNAL Timeout[stream, codes, msg]
};
$endOfStream => {
ERROR IO.EndOfStream[stream];
};
$networkStreamError => {
ERROR Error[stream, codes.rest, msg];
};
ENDCASE => {
ERROR IO.Error[ec, stream];
};
};
ENDCASE => {
ERROR IO.Error[ec, stream];
};
};
DeleteErrorDetails[d];
};
registrations: LIST OF NetworkStream.Registration Ź NIL;

Register: PUBLIC PROC [
protocolFamily: ATOM,
transportClass: ATOM,
proc: NetworkStream.RegistrationCallbackProc ]
~ {
prev: LIST OF NetworkStream.Registration;
each: LIST OF NetworkStream.Registration;
rNew: NetworkStream.Registration;
LookupForRegister: ENTRY PROC ~ {
ENABLE UNWIND => NULL;
prev Ź NIL; each Ź registrations;
DO
IF each = NIL THEN EXIT;
IF (each.first.protocolFamily = protocolFamily)
AND (each.first.transportClass = transportClass) THEN EXIT;
prev Ź each; each Ź each.rest;
ENDLOOP;
};
IF protocolFamily = NIL THEN RaiseIOError[Failure, NIL, LIST[$networkStreamError, $Register, $protocolFamily], "missing protocol family"];
IF transportClass = NIL THEN RaiseIOError[Failure, NIL, LIST[$networkStreamError, $Register, $transportClass], "missing transport class"];
LookupForRegister[];
SELECT TRUE FROM
(each = NIL ) => {
rNew Ź proc[NIL];
IF rNew # NIL
THEN registrations Ź CONS[
CompleteRegistration[protocolFamily, transportClass, rNew],
registrations];
};
(each.first.protocolFamily = protocolFamily)
AND (each.first.transportClass = transportClass) => {
rNew Ź proc[NIL];
SELECT TRUE FROM
rNew = each.first => NULL;
rNew # NIL => each.first Ź
CompleteRegistration[protocolFamily, transportClass, rNew];
prev = NIL => registrations Ź each.rest;
ENDCASE => prev.rest Ź each.rest;
};
ENDCASE => ERROR;
};

CompleteRegistration: PROC [family: ATOM, class: ATOM, r: NetworkStream.Registration]
RETURNS [NetworkStream.Registration]
~ {
insp: NetworkStream.NetworkStreamProcs Ź r.inNetworkStreamProcs;
onsp: NetworkStream.NetworkStreamProcs Ź r.outNetworkStreamProcs;
lp: NetworkStream.ListenerProcs Ź r.listenerProcs;

IF r.protocolFamily # family THEN RaiseIOError[Failure, NIL, LIST[$networkStreamError, $Register, $protocolFamily], "bad protocol family"];
IF r.transportClass # class THEN RaiseIOError[Failure, NIL, LIST[$networkStreamError, $Register, $transportClass], "bad transport class"];
IF (r.inStreamProcs = NIL) OR (insp = NIL) THEN RaiseIOError[Failure, NIL, LIST[$networkStreamError, $Register, $inputProcs], "bad input procs"];
IF (r.outStreamProcs = NIL) OR (onsp = NIL) THEN RaiseIOError[Failure, NIL, LIST[$networkStreamError, $Register, $outputProcs], "bad output procs"];
IF (r.createListener # NIL) AND (lp = NIL) THEN RaiseIOError[Failure, NIL, LIST[$networkStreamError, $Register, $listenerProcs], "bad listener procs"];

IF (r.createStreams = NIL) AND (r.createListener = NIL) THEN RaiseIOError[Failure, NIL, LIST[$networkStreamError, $Register, $create], "no create proc for streams or listener"];

IF r.createStreams = NIL THEN r.createStreams Ź DefaultRegisteredCreateStreamsProc;

IF insp.getStreamInfo = NIL THEN insp.getStreamInfo Ź DefaultGetStreamInfoProc;
IF insp.getTimeout = NIL THEN insp.getTimeout Ź DefaultGetTimeoutProc;
IF insp.setTimeout = NIL THEN insp.setTimeout Ź DefaultSetTimeoutProc;
IF insp.sendSoon = NIL THEN insp.sendSoon Ź InputSendSoonProc;
IF insp.getIndexDetails = NIL THEN RaiseIOError[Failure, NIL, LIST[$networkStreamError, $Register, $getIndexDetails], "no input GetIndexDetails proc"];
IF insp.getStreamState = NIL THEN insp.getStreamState Ź DefaultGetStreamStateProc;
IF insp.setSubStreamType = NIL THEN insp.setSubStreamType Ź InputSetSubStreamTypeProc;
IF insp.sendEndOfMessage = NIL THEN insp.sendEndOfMessage Ź InputSendEndOfMessageProc;
IF insp.waitAttention = NIL THEN insp.waitAttention Ź DefaultWaitAttentionProc;
IF insp.sendAttention = NIL THEN insp.sendAttention Ź InputSendAttentionProc;
IF insp.waitAck = NIL THEN insp.waitAck Ź InputWaitAckProc;
IF insp.finalizeStream = NIL THEN insp.finalizeStream Ź DefaultFinalizeStreamProc;

IF onsp.getStreamInfo = NIL THEN onsp.getStreamInfo Ź DefaultGetStreamInfoProc;
IF onsp.getTimeout = NIL THEN onsp.getTimeout Ź DefaultGetTimeoutProc;
IF onsp.setTimeout = NIL THEN onsp.setTimeout Ź DefaultSetTimeoutProc;
IF onsp.sendSoon = NIL THEN onsp.sendSoon Ź DefaultSendSoonProc;
IF onsp.getIndexDetails = NIL THEN RaiseIOError[Failure, NIL, LIST[$networkStreamError, $Register, $getIndexDetails], "no output GetIndexDetails proc"];
IF onsp.getStreamState = NIL THEN onsp.getStreamState Ź OutputGetStreamStateProc;
IF onsp.setSubStreamType = NIL THEN onsp.setSubStreamType Ź DefaultSetSubStreamTypeProc;
IF onsp.sendEndOfMessage = NIL THEN onsp.sendEndOfMessage Ź DefaultSendEndOfMessageProc;
IF onsp.waitAttention = NIL THEN onsp.waitAttention Ź OutputWaitAttentionProc;
IF onsp.sendAttention = NIL THEN onsp.sendAttention Ź DefaultSendAttentionProc;
IF onsp.waitAck = NIL THEN onsp.waitAck Ź DefaultWaitAckProc;
IF onsp.finalizeStream = NIL THEN onsp.finalizeStream Ź DefaultFinalizeStreamProc;

IF r.createListener = NIL THEN r.createListener Ź DefaultRegisteredCreateListenerProc;

IF lp # NIL THEN {
IF lp.getListenerInfo = NIL THEN lp.getListenerInfo Ź DefaultGetListenerInfoProc;
IF lp.destroyListener = NIL THEN RaiseIOError[Failure, NIL, LIST[$networkStreamError, $Register, $destroyListener], "bad destroyListener proc"];
IF lp.finalizeListener = NIL THEN lp.finalizeListener Ź DefaultFinalizeListenerProc;
};

RETURN [r];
};

Enumerate: PUBLIC ENTRY PROC [
families: ATOM,
classes: ATOM, 
proc: NetworkStream.EnumerateCallbackProc ]
~ {
FOR each: LIST OF NetworkStream.Registration Ź registrations, each.rest WHILE each # NIL DO
IF (families # NIL) AND (families # each.first.protocolFamily) THEN LOOP;
IF (classes # NIL) AND (classes # each.first.transportClass) THEN LOOP;
IF NOT proc[each.first.protocolFamily, each.first.transportClass] THEN EXIT;
ENDLOOP;
};

LookupRegistration: ENTRY PROC [protocolFamily: ATOM, transportClass: ATOM]
RETURNS [r: NetworkStream.Registration] ~ {
FOR each: LIST OF NetworkStream.Registration Ź registrations, each.rest WHILE each # NIL DO
IF (protocolFamily # NIL) AND (protocolFamily # each.first.protocolFamily) THEN LOOP;
IF (transportClass # NIL) AND (transportClass # each.first.transportClass) THEN LOOP;
RETURN [each.first];
ENDLOOP;
RETURN [NIL];
};
CreateUninitializedListener: PUBLIC PROC [registration: NetworkStream.Registration] RETURNS [listener: NetworkStream.Listener]
~ {
listener Ź NEW[NetworkStream.ListenerRecord Ź [
registration~registration,
procs~registration.listenerProcs
]];
};

CreateListener: PUBLIC PROC [protocolFamily: ATOM, local: ROPE, transportClass: ATOM, transportParameters: REF, listenerWorkerProc: NetworkStream.ListenerWorkerProc, listenerWorkerClientData: REF] RETURNS [listener: NetworkStream.Listener]
~ {
reg: NetworkStream.Registration;
reg Ź LookupRegistration[protocolFamily, transportClass];
IF reg = NIL THEN RaiseIOError[Failure, NIL, LIST[$networkStreamError, $notRegistered], "specified transport not registered"];
listener Ź reg.createListener[reg, local, transportParameters, listenerWorkerProc, listenerWorkerClientData];
[] Ź FinalizeOps.EnableFinalization[listener, theFQ];
};

DefaultRegisteredCreateListenerProc: NetworkStream.RegisteredCreateListenerProc ~ {
RaiseIOError[Failure, NIL, LIST[$networkStreamError, $InitListener, $notImplemented], "listeners not implemented"];
};

DefaultGetListenerInfoProc: NetworkStream.GetListenerInfoProc -- [listener] RETURNS [protocolFamily, local, transportClass, proc, clientData] -- ~ {
r: NetworkStream.Registration Ź listener.registration;
protocolFamily Ź r.protocolFamily;
local Ź listener.local;
transportClass Ź r.transportClass;
proc Ź listener.listenerWorkerProc;
clientData Ź listener.listenerWorkerClientData;
};

DefaultFinalizeListenerProc: NetworkStream.FinalizeListenerProc ~ {
ENABLE IO.Error, IO.EndOfStream, Error, Timeout => CONTINUE;
NetworkStream.DestroyListener[listener];
};

CreateUninitializedStreams: PUBLIC PROC [registration: NetworkStream.Registration] RETURNS [in: IO.STREAM, out: IO.STREAM] ~ {
inNetworkStreamData: NetworkStream.NetworkStreamData;
outNetworkStreamData: NetworkStream.NetworkStreamData;
inNetworkStreamData Ź NEW[NetworkStream.NetworkStreamDataRecord Ź [registration~registration, procs~registration.inNetworkStreamProcs]];
in Ź IO.CreateStream[streamProcs~registration.inStreamProcs, streamData~inNetworkStreamData];
outNetworkStreamData Ź NEW[NetworkStream.NetworkStreamDataRecord Ź [registration~registration, procs~registration.outNetworkStreamProcs]];
out Ź IO.CreateStream[streamProcs~registration.outStreamProcs, streamData~outNetworkStreamData];
};

CreateStreams: PUBLIC PROC [protocolFamily: ATOM, remote: ROPE, transportClass: ATOM, timeout: NetworkStream.Milliseconds, transportParameters: REF] RETURNS [in: IO.STREAM, out: IO.STREAM] ~ {
reg: NetworkStream.Registration;
reg Ź LookupRegistration[protocolFamily, transportClass];
IF reg = NIL THEN RaiseIOError[Failure, NIL, LIST[$networkStreamError, $notRegistered], "specified transport not registered"];
[in, out] Ź reg.createStreams[reg, remote, timeout, transportParameters];
[] Ź FinalizeOps.EnableFinalization[in, theFQ];
[] Ź FinalizeOps.EnableFinalization[out, theFQ];
};

DefaultRegisteredCreateStreamsProc: NetworkStream.RegisteredCreateStreamsProc ~ {
RaiseIOError[Failure, NIL, LIST[$networkStreamError, $CreateStreams, $notImplemented], "connect not implemented"];
};
DefaultGetStreamInfoProc: NetworkStream.GetStreamInfoProc ~ {
d: NetworkStream.NetworkStreamData Ź NARROW[stream.streamData];
RETURN [protocolFamily~d.registration.protocolFamily, local~d.local, remote~d.remote, transportClass~d.registration.transportClass] };

DefaultGetTimeoutProc: NetworkStream.GetTimeoutProc ~ {
d: NetworkStream.NetworkStreamData Ź NARROW[stream.streamData];
RETURN [timeout~d.timeout, signalTimeout~d.signalTimeout] };

DefaultSetTimeoutProc: NetworkStream.SetTimeoutProc ~ {
IF timeout # NetworkStream.waitForever
THEN RaiseIOError[NotImplementedForThisStream, stream, LIST[$notImplemented, $SetTimeout], "SetTimeout not implemented"];
};

DefaultGetStreamStateProc: NetworkStream.GetStreamStateProc ~ {
IF reset THEN RaiseIOError[NotImplementedForThisStream, in, LIST[$notImplemented, $GetStreamState], "reset stream state not implemented"];
RETURN [streamState~open, subStreamType~0, attentionType~0];
};

OutputGetStreamStateProc: NetworkStream.GetStreamStateProc ~ {
RaiseIOError[NotImplementedForThisStream, in, LIST[$notImplemented, $GetStreamState], "GetStreamState not implemented for output stream"];
ERROR;
};

DefaultSetSubStreamTypeProc: NetworkStream.SetSubStreamTypeProc ~ {
IF subStreamType # 0 THEN RaiseIOError[NotImplementedForThisStream, out, LIST[$notImplemented, $SetSubStreamType], "SetSubStreamType not implemented"];
};

InputSetSubStreamTypeProc: NetworkStream.SetSubStreamTypeProc ~ {
RaiseIOError[NotImplementedForThisStream, out, LIST[$notImplemented, $SetSubStreamType], "SetSubStreamType not implemented for input stream"];
};

DefaultSendEndOfMessageProc: NetworkStream.SendEndOfMessageProc ~ {
RaiseIOError[NotImplementedForThisStream, out, LIST[$notImplemented, $SendEndOfMessage], "SendEndOfMessage not implemented"];
};

InputSendEndOfMessageProc: NetworkStream.SendEndOfMessageProc ~ {
RaiseIOError[NotImplementedForThisStream, out, LIST[$notImplemented, $SendEndOfMessage], "SendEndOfMessage not implemented for input stream"];
};

DefaultWaitAttentionProc: NetworkStream.WaitAttentionProc ~ {
EntryWaitAttention: ENTRY PROC ~ {
ENABLE UNWIND => NULL;
attn: CONDITION;
TRUSTED {
Process.EnableAborts[@attn];
IF timeout = NetworkStream.waitForever
THEN { Process.DisableTimeout[@attn] }
ELSE {  Process.SetTimeout[@attn, Process.MsecToTicks[timeout]] };
};
WAIT attn;
};
DO
EntryWaitAttention[];
RaiseIOError[Failure, in, LIST[$timeout, $waitAttention], "WaitAttention timeout"];
ENDLOOP;
};

OutputWaitAttentionProc: NetworkStream.WaitAttentionProc ~ {
RaiseIOError[NotImplementedForThisStream, in, LIST[$notImplemented, $WaitAttention], "WaitAttention not implemented for output stream"];
RETURN [0]
};

DefaultSendAttentionProc: NetworkStream.SendAttentionProc ~ {
RaiseIOError[NotImplementedForThisStream, out, LIST[$notImplemented, $SendAttention], "SendAttention not implemented"];
};

InputSendAttentionProc: NetworkStream.SendAttentionProc ~ {
RaiseIOError[NotImplementedForThisStream, out, LIST[$notImplemented, $SendAttention], "SendAttention not implemented for input stream"];
};

InputWaitAckProc: NetworkStream.WaitAckProc ~ {
RaiseIOError[NotImplementedForThisStream, out, LIST[$notImplemented, $WaitAck], "WaitAck not implemented for input stream"];
};

DefaultWaitAckProc: NetworkStream.WaitAckProc ~ {
ackIndex, bufferIndex: INT;
IF index = 0 THEN RETURN;
[ackIndex~ackIndex, bufferIndex~bufferIndex] Ź NetworkStream.GetIndexDetails[out];
IF index < 0 THEN index Ź bufferIndex;
IF ackIndex >= index THEN RETURN;
RaiseIOError[NotImplementedForThisStream, out, LIST[$notImplemented, $WaitAck], "WaitAck not implemented"];
};

DefaultFinalizeStreamProc: NetworkStream.FinalizeStreamProc ~ {
ENABLE IO.Error, IO.EndOfStream, Error, Timeout => CONTINUE;
IO.Close[stream];
};


sendSoonCommTimer: CommTimer.Timer;

sendSoonTail: STREAM Ź NIL;
sendSoonListNotEmpty: CONDITION;

GetFromSendSoonList: ENTRY PROC RETURNS [s: STREAM, sendSoonIndex: INT] ~ {
ENABLE UNWIND => NULL;
tailData, sData: NetworkStream.NetworkStreamData;
WHILE sendSoonTail = NIL DO WAIT sendSoonListNotEmpty ENDLOOP;
tailData Ź NARROW[sendSoonTail.streamData];
s Ź tailData.sendSoonNext;
sData Ź NARROW[s.streamData];
IF s = sendSoonTail
THEN sendSoonTail Ź NIL
ELSE tailData.sendSoonNext Ź sData.sendSoonNext;
sendSoonIndex Ź sData.sendSoonIndex;
sData.sendSoonIndex Ź -1;
sData.sendSoonNext Ź NIL;
};

AddToSendSoonList: ENTRY PROC [s: STREAM] ~ {
ENABLE UNWIND => NULL;
sData, tailData: NetworkStream.NetworkStreamData;
sData Ź NARROW[s.streamData];
IF sData.sendSoonNext # NIL THEN RETURN;
IF sendSoonTail = NIL
THEN {
sendSoonTail Ź sData.sendSoonNext Ź s;
}
ELSE {
tailData Ź NARROW[sendSoonTail.streamData];
sData.sendSoonNext Ź tailData.sendSoonNext;
sendSoonTail Ź tailData.sendSoonNext Ź s;
};
NOTIFY sendSoonListNotEmpty;
};

SendSoonEventProc: CommTimer.EventProc -- [clientData] -- ~ {
AddToSendSoonList[NARROW[clientData]];
};

SendSoonDaemon: PROC ~ {
s: STREAM;
sendSoonIndex: INT;
DO
[s, sendSoonIndex] Ź GetFromSendSoonList[];
{ENABLE IO.Error, Error, Timeout, ABORTED => CONTINUE;
IF NetworkStream.GetIndexDetails[s].bufferIndex < sendSoonIndex
THEN IO.Flush[s];
};
ENDLOOP;
};

DefaultSendSoonProc: NetworkStream.SendSoonProc ~ {
d: NetworkStream.NetworkStreamData;
oldIndex: INT Ź 0;
EntrySwapIndex: ENTRY PROC ~ {
oldIndex Ź d.sendSoonIndex;
d.sendSoonIndex Ź IO.GetIndex[out];
};
IF when = 0 THEN { IO.Flush[out]; RETURN };
d Ź NARROW[out.streamData];
EntrySwapIndex[];
IF oldIndex < 0 THEN [] Ź CommTimer.ScheduleEvent[sendSoonCommTimer, when, SendSoonEventProc, out];
};

InputSendSoonProc: NetworkStream.SendSoonProc ~ {
RaiseIOError[NotImplementedForThisStream, out, LIST[$notImplemented, $SendSoon], "SendSoon not implemented for input stream"];
};

StartCodeForSendSoon: PROC ~ {
sendSoonCommTimer Ź CommTimer.CreateTimer[sendSoonGrainSizeMsec, sendSoonExpectedWaitMsec];
TRUSTED { Process.Detach[ FORK SendSoonDaemon[] ] };
};
theFQ: FinalizeOps.CallQueue Ź FinalizeOps.CreateCallQueue[Finalizer];

Finalizer: FinalizeOps.FinalizeProc ~ {
ENABLE ABORTED => CONTINUE;
WITH object SELECT FROM
s: IO.STREAM => {
d: NetworkStream.NetworkStreamData Ź NARROW[s.streamData];
d.procs.finalizeStream[s];
};
l: NetworkStream.Listener => {
l.procs.finalizeListener[l];
};
ENDCASE => ERROR;
};

debugMsgs: BOOL Ź FALSE;

SetDebugMsgs: PROC [i: INT] RETURNS [prev: INT] ~ {
prev Ź IF debugMsgs THEN 1 ELSE 0;
debugMsgs Ź (i # 0);
};

DBMsgInner: PROC [fmt, s: CStrings.CString] ~ TRUSTED MACHINE CODE { "XR_PrintF" };

debugFmt: CStrings.CString Ź UXStrings.Create["%s"];

DBMsg: PROC [r: ROPE] ~ {
s: CStrings.CString Ź UXStrings.Create[r];
DBMsgInner[debugFmt, s];
};

DBPutErr: PROC [m1: ROPE, codes: LIST OF ATOM, m2: ROPE] ~ {
DBMsg[m1]; DBMsg[" codes: "];
FOR each: LIST OF ATOM Ź codes, each.rest WHILE each # NIL DO
DBMsg[Convert.RopeFromAtom[each.first]];
DBMsg[" "];
ENDLOOP;
DBMsg[", msg: "];  DBMsg[m2];  DBMsg["\n"];
};

DBAtomFromString: PROC [s: CStrings.CString] RETURNS [ATOM] ~ {
r: ROPE Ź UXStrings.ToRope[s, 100];
RETURN [Convert.AtomFromRope[r]];
};

DBEchoWorker: NetworkStream.ListenerWorkerProc -- [listener, in, out] -- ~ {
ENABLE UNWIND => { IO.Close[in, TRUE]; IO.Close[out, TRUE] };
c: CHAR;
DO
c Ź IO.GetChar[in ! IO.EndOfStream => EXIT];
IO.PutChar[out, c];
NetworkStream.SendSoon[out, 0];
ENDLOOP;
IO.Close[in, TRUE]; IO.Close[out, TRUE]; 
};

DBStartEchoer: PROC [pf: ATOM, local: ROPE, tc: ATOM]
RETURNS [listener: NetworkStream.Listener] ~ {
listener Ź NetworkStream.CreateListener[pf, local, tc, NIL, DBEchoWorker, NIL];
};

DBComplexEchoWorker: NetworkStream.ListenerWorkerProc -- [listener, in, out] -- ~ {
ENABLE UNWIND => { IO.Close[in, TRUE]; IO.Close[out, TRUE] };
c: CHAR;
atEndOfStream: BOOL Ź FALSE;
streamState: NetworkStream.StreamState;
subStreamType: NetworkStream.SubStreamType;
attentionType: NetworkStream.AttentionType;
errorKind: CARD Ź 0;
errorCodes: LIST OF ATOM;
errorMsg: ROPE;
IF debugMsgs THEN DBMsg["Echoer entered\n"];
DO
IF atEndOfStream THEN {
[streamState, subStreamType, attentionType] Ź
NetworkStream.GetStreamState[in, TRUE];
IF debugMsgs THEN {
m: ROPE Ź IO.PutFR["Echoer atEndOfStream, state %g sst %g attn %g\n", IO.card[ORD[streamState]], IO.card[subStreamType], IO.card[attentionType]];
DBMsg[m];
};
IF streamState = remoteClosed THEN EXIT;
atEndOfStream Ź FALSE;
LOOP;
};
IF errorKind # 0 THEN {
IF debugMsgs THEN {
DBPutErr[IO.PutFR1["Echoer err %g ", IO.card[errorKind]], errorCodes, errorMsg];
};
EXIT;
};
{
ENABLE {
NetworkStream.Error => {
errorKind Ź 1;  errorCodes Ź codes;  errorMsg Ź msg;
LOOP };
NetworkStream.Timeout => {
errorKind Ź 2;  errorCodes Ź codes;  errorMsg Ź msg;
LOOP };
IO.Error => {
errorKind Ź 3;  [codes~errorCodes, msg~errorMsg] Ź NetworkStream.GetIOErrorDetails[stream];
LOOP };
IO.EndOfStream => {
atEndOfStream Ź TRUE;
LOOP };
};
c Ź IO.GetChar[in];
};
IF debugMsgs THEN {
m: ROPE Ź IO.PutFR1["Echoer got %g, replying ...\n", IO.char[c]];
DBMsg[m];
};
IO.PutChar[out, c];
IF debugMsgs THEN {
DBMsg["echoer done putchar ...\n"];
};
NetworkStream.SendSoon[out, 0];
IF debugMsgs THEN {
DBMsg["echoer done SendSoon.\n"];
};
ENDLOOP;
IF debugMsgs THEN {
DBMsg["Echoer exit\n"];
};
IO.Close[in, TRUE]; IO.Close[out, TRUE]; 
};

DBStartComplexEchoer: PROC [pf: ATOM, local: ROPE, tc: ATOM]
RETURNS [listener: NetworkStream.Listener] ~ {
listener Ź NetworkStream.CreateListener[pf, local, tc, NIL, DBComplexEchoWorker, NIL];
};

DBListenerInfo: TYPE ~ RECORD [pf: ATOM, local: ROPE, tc: ATOM];

DBGetListenerInfo: PROC [l: NetworkStream.Listener] RETURNS [info: REF DBListenerInfo] ~ {
info Ź NEW[DBListenerInfo Ź [NIL, NIL, NIL]];
[protocolFamily~info.pf, local~info.local, transportClass~info.tc] Ź NetworkStream.GetListenerInfo[l];
};

DBStreamInfo: TYPE ~ RECORD [pf: ATOM, local: ROPE, remote: ROPE, tc: ATOM];

DBGetStreamInfo: PROC [s: IO.STREAM] RETURNS [info: REF DBStreamInfo] ~ {
info Ź NEW[DBStreamInfo Ź [NIL, NIL, NIL, NIL]];
[protocolFamily~info.pf, local~info.local, remote~info.remote, transportClass~info.tc] Ź NetworkStream.GetStreamInfo[s];
};

DBStreamPair: TYPE ~ RECORD [ in, out: IO.STREAM ];

DBCreateStreamPair: PROC [pf: ATOM, remote: ROPE, tc: ATOM, timeoutMsec: CARD32] RETURNS [streamPair: REF DBStreamPair] ~ {
streamPair Ź NEW[DBStreamPair Ź [NIL, NIL]];
[streamPair.in, streamPair.out] Ź NetworkStream.CreateStreams[pf, remote, tc, timeoutMsec, NIL];
};

DBCallProtected: PROC [callee: PROC] ~ {
ENABLE {
NetworkStream.Error => {
IF debugMsgs THEN {
DBPutErr["NS.Err", codes, msg];
CONTINUE;
};
REJECT;
};
NetworkStream.Timeout => {
IF debugMsgs THEN {
DBPutErr["NS.Timeout", codes, msg];
CONTINUE;
};
REJECT;
};
IO.Error => {
IF debugMsgs THEN {
codes: LIST OF ATOM;
msg: ROPE;
[codes, msg] Ź NetworkStream.GetIOErrorDetails[stream];
DBPutErr["IO.Err", codes, msg];
CONTINUE;
};
REJECT;
};
IO.EndOfStream => {
IF debugMsgs THEN {
DBPutErr["IO.EndOfStream", NIL, NIL];
CONTINUE;
};
REJECT;
};
};
callee[];
};

DBPutC: PROC [s: IO.STREAM, c: CHAR] ~ {
DoIt: PROC ~ { IO.PutChar[s, c] };
DBCallProtected[DoIt];
};

DBFlushC: PROC [s: IO.STREAM] ~ {
DoIt: PROC ~ { IO.Flush[s] };
DBCallProtected[DoIt];
};

DBGetC: PROC [s: IO.STREAM] RETURNS[c: CHAR] ~ {
DoIt: PROC ~ { c Ź IO.GetChar[s] };
c Ź '?;
DBCallProtected[DoIt];
};

DBGetBlock: PROC [s: IO.STREAM, n: INT] RETURNS [b: REF TEXT] ~ {
DoIt: PROC ~ {
nGot: INT;
b Ź RefText.New[n];
nGot Ź IO.GetBlock[s, b, 0, n];
IF (nGot < n) AND (debugMsgs) THEN {
DBMsg[IO.PutFR["GetBlock got only %g of %g\n", IO.int[nGot], IO.int[n]]];
};
};
DBCallProtected[DoIt];
};

DBPutBlock: PROC [s: IO.STREAM, n: INT, c: CHAR] ~ {
DoIt: PROC ~ {
b: REF TEXT Ź RefText.New[n];
FOR i: INT IN [0..n) DO
b[i] Ź c;
ENDLOOP;
IO.PutBlock[s, b, 0, n];
};
DBCallProtected[DoIt];
};
StartCodeForSendSoon[];

}.
     
NetworkStreamImpl.mesa
Copyright Ó 1989, 1991, 1992 by Xerox Corporation.  All rights reserved.
Demers, May 7, 1990 9:21 am PDT
Last tweaked by Mike Spreitzer on November 9, 1989 4:36:12 pm PST
Willie-s, September 30, 1991 1:49 pm PDT
Michael Plass, November 22, 1991 4:21 pm PST
Christian Jacobi, July 24, 1992 2:05 pm PDT
Types
Parameters
Public Errors
Error Reporting Implementation 
Registration
Listener Creation
Stream Creation
Default Stream Procs
DefaultSendSoonProc: NetworkStream.FinalizeListenerProc ~ { (see below) };
InputSendSoonProc: NetworkStream.FinalizeListenerProc ~ { (see below) };
SendSoon Implementation
Invariant:
d.sendSoonIndex >= 0 ==> eventually an IO.Flush will be performed to flush beyond that index value.
d.sendSoonIndex < 0 ==> stream is not on sendSoonList.
THIS SHOULD BE FIXED TO HAVE A SEPARATE PROCESS PER STREAM - ajd.
FinalizeOps
Debugging Stuff
IF streamState = open THEN
Start
 Ęe  (cedarcode) styleNewlineDelimiter
 codeK
Ďe=HKKAK(K,K+K Ďk		K
K	ĎcK	KKKKK	KK
KK Ďn K",WKKheadKK
K K% 		HK  		KK  2K.K K,#KK	KK	KK 15))KK	KKK  -dK,KKKKKKK  )GKKKKKKKKKKDK	KK  &	pKKK-:K,KKK,K;KKKK  		^K##K	DK%%KK  	`K##K
FK+KK	/K
7K"KKKK %KKKKKKKKK8K 
 0rKK)K)K!! !KK!KK..kKKKK
NK
NKKKMfK-2bKKKLVK(K!KKKK 
 
	!zKK@@KAA22K KJKJK	BK	DKHK KUąK K6SK K/OK)FK)FK#>K
UK1RK3VK3VK/OK-MK!;K1RK K/OK)FK)FK%@K
VK0QK5XK5XK.NK/OK#=K1RK K8VK K1QK
PK3TKK KKK 
 	.jK7[K
(IK
'GK<LKKK  $w7[K
.UK
.UKKKK ,#~K!/KK  KKK  	R#ďKK  K99K
M~KmmK5KK  #0SKTsKK  Đbn$RK6K""KK""K##K//KK  (CK <K((K  ,		~K55K66KoKV]KpKX`KK  
<ŔK  K99K
M~KIIK/K0KK  "/QKSrK %=K%?KK  "7K%?K6<K  "7$&K3>yKK  &?K/JK6<KK  &>K.XKKK  (CK0JKK  (AK/[KK  (CK/J}KK  (AK/[KK  %= "KK		K$&K"&K>BKK
KKK5SKKK  %<K.VK
KK  %=K/DwKK  %;K/UKK  /K/I|KK  1KKKRRK&K!K/8kKK  &?K <KKK K 7JK 7H
KcK6AK K##K KK	 K  KKK11K
>K+KKKK,0K$$KKKK  -KK11KK(K&&KK+K++K))KKKK  =K&KK  K
K++K	6=?K
KKKK   3K##K
 KK#KK
+KKKOcK   1K/K~KK  K[[K4KKFK  	'KK%:KKKKKKK KK 
 3K"KKK K
 
SK K5 K*KKK  	<K=K(KKK+KK  ?K#K!KK  #LK=KK,KKKK)KK  	'dK7OKK  #SK=KKK'K++K++KKK
K,KOUK:K	KK(KKKK	)PKKKK44KK44KK[KKKKKKK)
AK	KKK#KKK!!KKKKK)KK  	'kK7VKK 	@K  ZK	-KffKK 	
LK  IK0KxxKK 	3K  
{K,K[`KK  
(KK	KKKK##K	KKKKK
K77KK	KKKK%K	KKKKK	KK 
 (K "KKK  !K KKK  0K 	#KKKK  
A K
KK$K'
IKKKKK  
4 KK	KKKKKKK K     ]  }  