DIRECTORY
CoFork USING[CoForkMe],
RefAnyStream USING[Handle, Object, ObjectProcs, ErrorCode];

RefAnyStreamImpl: MONITOR  LOCKS NARROW[self.data, StreamData] USING self: RefAnyStream.Handle
IMPORTS CoFork
EXPORTS RefAnyStream =
BEGIN





Error: PUBLIC ERROR [ec: RefAnyStream.ErrorCode] = CODE;




RefAnySource: TYPE = POINTER TO PORT[continue: BOOLEAN, aItem: REF ANY] RETURNS[more: BOOLEAN, bItem: REF ANY]; -- used by client
RefAnySink: TYPE = POINTER TO PORT[more: BOOLEAN, bItem: REF ANY] RETURNS[continue: BOOLEAN, aItem: REF ANY]; -- used by co-fork
RefAnyStopper: TYPE = PROCEDURE[continue: BOOLEAN, aItem: REF ANY]; -- used by client


StreamData: TYPE = REF StreamDataBody;

StreamDataBody: TYPE = MONITORED RECORD[
source: RefAnySource,
stop: RefAnyStopper,

itemReady: BOOLEAN,
item: REF ANY,

sink: RefAnySink];





CoForkedProducerStreamProcs: REF RefAnyStream.ObjectProcs _ NEW[RefAnyStream.ObjectProcs _ [get: CoForkedProducerGet, put: CoForkedProducerPut, putBack:  CoForkedProducerPutBack, flush: CoForkedProducerFlush, close: CoForkedProducerClose, endOf: CoForkedProducerEndOf, empty: ErrProc]];

ErrProc: PROCEDURE[self: RefAnyStream.Handle] RETURNS[BOOLEAN] =
{ERROR Error[NotImplementedForThisStream]};

CoForkedProducerGet: ENTRY PROCEDURE[self: RefAnyStream.Handle] RETURNS[REF ANY] =
{ENABLE UNWIND => NULL; ERROR Error[NotImplementedForThisStream]};

CoForkedProducerPut: ENTRY PROCEDURE[self: RefAnyStream.Handle, event: REF ANY] =
BEGIN ENABLE UNWIND => NULL; 
d: StreamData _ NARROW[self.data];
IF d.sink # NIL THEN {IF NOT (d.sink[TRUE, event]).continue THEN d.sink _ NIL} ELSE ERROR Error[StreamClosed];
END;

CoForkedProducerPutBack: ENTRY PROCEDURE[self: RefAnyStream.Handle, event: REF ANY] =
{ENABLE UNWIND => NULL; ERROR Error[NotImplementedForThisStream]};

CoForkedProducerFlush: ENTRY PROCEDURE[self: RefAnyStream.Handle] =
{ENABLE UNWIND => NULL; ERROR Error[NotImplementedForThisStream]};

CoForkedProducerClose: ENTRY PROCEDURE[self: RefAnyStream.Handle] =
BEGIN ENABLE UNWIND => NULL; 
d: StreamData _ NARROW[self.data];
d.sink _ NIL;
END;

CoForkedProducerEndOf: ENTRY PROCEDURE[self: RefAnyStream.Handle] RETURNS[BOOLEAN] =
BEGIN ENABLE UNWIND => NULL; 
d: StreamData _ NARROW[self.data];
RETURN[d.sink=NIL];
END;





ClientConsumerStreamProcs: REF RefAnyStream.ObjectProcs _ NEW[RefAnyStream.ObjectProcs _ [get: ClientConsumerGet, put: ClientConsumerPut, putBack: ClientConsumerPutBack, flush: ClientConsumerFlush, close: ClientConsumerClose, endOf: ClientConsumerEndOf, empty: ErrProc]];

ClientConsumerGet: ENTRY PROCEDURE[self: RefAnyStream.Handle] RETURNS[REF ANY] =
BEGIN ENABLE UNWIND => NULL; 
d: StreamData _ NARROW[self.data];
IF d.source = NIL THEN ERROR Error[StreamClosed];
IF d.itemReady THEN {d.itemReady _ FALSE; RETURN[d.item]};
[d.itemReady, d.item] _ d.source[TRUE, NIL];
IF d.itemReady
THEN {d.itemReady _ FALSE; RETURN[d.item]}
ELSE {d.stop[FALSE, NIL];
d.source _ NIL;
ERROR Error[StreamClosed]};
END;

ClientConsumerPut: ENTRY PROCEDURE[self: RefAnyStream.Handle, event: REF ANY] =
{ENABLE UNWIND => NULL; ERROR Error[NotImplementedForThisStream]};

ClientConsumerPutBack: ENTRY PROCEDURE[self: RefAnyStream.Handle, event: REF ANY] =
{ENABLE UNWIND => NULL; ERROR Error[NotImplementedForThisStream]}; -- not implemented

ClientConsumerFlush: ENTRY PROCEDURE[self: RefAnyStream.Handle] =
{ENABLE UNWIND => NULL; ERROR Error[NotImplementedForThisStream]};

ClientConsumerEndOf: ENTRY PROCEDURE[self: RefAnyStream.Handle] RETURNS[BOOLEAN] =
BEGIN ENABLE UNWIND => NULL; 
d: StreamData _ NARROW[self.data];
IF d.itemReady THEN RETURN[FALSE];
IF d.source = NIL THEN RETURN[TRUE];
[d.itemReady, d.item] _ d.source[TRUE, NIL];
RETURN[NOT d.itemReady]
END;

ClientConsumerClose: ENTRY PROCEDURE[self: RefAnyStream.Handle] =
BEGIN ENABLE UNWIND => NULL; 
d: StreamData _ NARROW[self.data];
IF d.source # NIL
THEN {d.stop[FALSE, NIL];
d.source _ NIL};
END;





MakeCoForkedProducer: PROCEDURE[producer: PROC[REF ANY, RefAnyStream.Handle], data: REF ANY] RETURNS[genItem: RefAnySource, stopItems: RefAnyStopper] =
BEGIN
sink: RefAnySink _ CoFork.CoForkMe[];
d: StreamData _ NEW[StreamDataBody _ [ , NIL, NIL, FALSE, NIL, sink]];
sinkStream: RefAnyStream.Handle;
IF NOT (sink[FALSE, NIL]).continue THEN d.sink _ NIL;
sinkStream _ NEW[RefAnyStream.Object _ [procs: CoForkedProducerStreamProcs, data: d]];
producer[data, sinkStream];
END;

CreateCoForkedProducerStream: PUBLIC PROC[producer: PROC[self: REF ANY, sink: RefAnyStream.Handle--for use by the co-forked producer--], data: REF ANY]
RETURNS[RefAnyStream.Handle--for use by the client consumer--] =
BEGIN
source: RefAnySource;
stop: RefAnyStopper;
d: StreamData;
sourceStream: RefAnyStream.Handle;
[source, stop] _ MakeCoForkedProducer[producer, data];
d _ NEW[StreamDataBody _ [ , source, stop, FALSE, NIL, NIL]];
sourceStream _ NEW[RefAnyStream.Object _ [procs: ClientConsumerStreamProcs, data: d]];
RETURN[sourceStream];
END;





ClientProducerStreamProcs: REF RefAnyStream.ObjectProcs _ NEW[RefAnyStream.ObjectProcs _ [get: ClientProducerGet, put: ClientProducerPut, putBack:  ClientProducerPutBack, flush: ClientProducerFlush, close: ClientProducerClose, endOf: ClientProducerEndOf, empty: ErrProc]];


ClientProducerGet: ENTRY PROCEDURE[self: RefAnyStream.Handle] RETURNS[REF ANY] =
{ENABLE UNWIND => NULL; ERROR Error[NotImplementedForThisStream]};

ClientProducerPut: ENTRY PROCEDURE[self: RefAnyStream.Handle, event: REF ANY] =
BEGIN ENABLE UNWIND => NULL; 
d: StreamData _ NARROW[self.data];
IF d.source # NIL THEN {IF NOT (d.source[TRUE, event]).more THEN d.source _ NIL} ELSE ERROR Error[StreamClosed];
END;

ClientProducerPutBack: ENTRY PROCEDURE[self: RefAnyStream.Handle, event: REF ANY] =
{ENABLE UNWIND => NULL; ERROR Error[NotImplementedForThisStream]};

ClientProducerFlush: ENTRY PROCEDURE[self: RefAnyStream.Handle] =
{ENABLE UNWIND => NULL; ERROR Error[NotImplementedForThisStream]};

ClientProducerClose: ENTRY PROCEDURE[self: RefAnyStream.Handle] =
BEGIN ENABLE UNWIND => NULL; 
d: StreamData _ NARROW[self.data];
IF d.source # NIL
THEN {d.stop[FALSE, NIL];
d.source _ NIL};
END;

ClientProducerEndOf: ENTRY PROCEDURE[self: RefAnyStream.Handle] RETURNS[BOOLEAN] =
BEGIN ENABLE UNWIND => NULL; 
d: StreamData _ NARROW[self.data];
RETURN[d.source=NIL];
END;





CoForkedConsumerStreamProcs: REF RefAnyStream.ObjectProcs _ NEW[RefAnyStream.ObjectProcs _ [get: CoForkedConsumerGet, put: CoForkedConsumerPut, putBack: CoForkedConsumerPutBack, flush: CoForkedConsumerFlush, close: CoForkedConsumerClose, endOf: CoForkedConsumerEndOf, empty: ErrProc]];

CoForkedConsumerGet: ENTRY PROCEDURE[self: RefAnyStream.Handle] RETURNS[REF ANY] =
BEGIN ENABLE UNWIND => NULL; 
d: StreamData _ NARROW[self.data];
IF d.sink = NIL THEN ERROR Error[StreamClosed];
IF d.itemReady THEN {d.itemReady _ FALSE; RETURN[d.item]};
[d.itemReady, d.item] _ d.sink[TRUE, NIL];
IF d.itemReady
THEN {d.itemReady _ FALSE; RETURN[d.item]}
ELSE {d.sink _ NIL; ERROR Error[StreamClosed]};
END;

CoForkedConsumerPut: ENTRY PROCEDURE[self: RefAnyStream.Handle, event: REF ANY] =
{ENABLE UNWIND => NULL; ERROR Error[NotImplementedForThisStream]};

CoForkedConsumerPutBack: ENTRY PROCEDURE[self: RefAnyStream.Handle, event: REF ANY] =
{ENABLE UNWIND => NULL; ERROR Error[NotImplementedForThisStream]}; -- not implemented

CoForkedConsumerFlush: ENTRY PROCEDURE[self: RefAnyStream.Handle] =
{ENABLE UNWIND => NULL; ERROR Error[NotImplementedForThisStream]};

CoForkedConsumerEndOf: ENTRY PROCEDURE[self: RefAnyStream.Handle] RETURNS[BOOLEAN] =
BEGIN ENABLE UNWIND => NULL; 
d: StreamData _ NARROW[self.data];
IF d.itemReady THEN RETURN[FALSE];
IF d.sink = NIL THEN RETURN[TRUE];
[d.itemReady, d.item] _ d.source[TRUE, NIL];
IF NOT d.itemReady THEN d.sink _ NIL;
RETURN[NOT d.itemReady]
END;

CoForkedConsumerClose: ENTRY PROCEDURE[self: RefAnyStream.Handle] =
BEGIN ENABLE UNWIND => NULL; 
d: StreamData _ NARROW[self.data];
d.sink _ NIL;
END;



MakeCoForkedConsumer: PROCEDURE[consumer: PROC[REF ANY, RefAnyStream.Handle], data: REF ANY] RETURNS[genItem: RefAnySource, stopItems: RefAnyStopper] =
BEGIN
sink: RefAnySink _ CoFork.CoForkMe[];
d: StreamData _ NEW[StreamDataBody _ [ , NIL, NIL, FALSE, NIL, sink]];
sinkStream: RefAnyStream.Handle;
IF NOT (sink[FALSE, NIL]).continue THEN d.sink _ NIL;
sinkStream _ NEW[RefAnyStream.Object _ [procs: CoForkedConsumerStreamProcs, data: d]];
consumer[data, sinkStream];
END;

CreateCoForkedConsumerStream: PUBLIC PROC[consumer: PROC[self: REF ANY, source: RefAnyStream.Handle--for use by the co-forked consumer--], data: REF ANY]
RETURNS[RefAnyStream.Handle--for use by the client producer--] =
BEGIN
source: RefAnySource;
stop: RefAnyStopper;
d: StreamData;
sourceStream: RefAnyStream.Handle;
[source, stop] _ MakeCoForkedConsumer[consumer, data];
d _ NEW[StreamDataBody _ [ , source, stop, FALSE, NIL, NIL]];
sourceStream _ NEW[RefAnyStream.Object _ [procs: ClientProducerStreamProcs, data: d]];
RETURN[sourceStream];
END;




END.


ZRefAnyStreamImpl.mesa
Last Modified On 22-Dec-81 14:30:06 By Sturgis
Last Modified On 21-Jan-82 12:02:44 By Paul Rovner
nomenclature
sink is associated with the co-forked procedure
source, stopper are associated with the client
producer puts ref-anys
consumer gets ref-anys
co-fork refers to the "co-forked" call stack, this side signals end by returning
client refers to the non "co-forked" code, this side signals end by calling stopper.
if the producer is co-forked, then the producer will put ref-anys to the sink
it the producer is the client, then the producer will put ref-anys to the source
errors
port declarations, must follow a precise pattern specified by CoFork
aItems go from client to co-fork, bItems go from co-fork to client
the following record type is used by all stream types so that I can compile all in one module, so that the locks expression is uniform for all entry procedures  (ugh!!!)  note that some fields are unused in each case.
following fields are used in the client stream
following fields are used by the consumer
following fields are used in the co-forked stream
RefAnyStream implementation seen by a co-forked RefAny producer
RefAnyStream implementation seen by a client consumer
co-forked producer client code
RefAnyStream implementation seen by a client RefAny producer
RefAnyStream implementation seen by a co-forked consumer
co-forked consumer client code
16-Dec-81 17:26:53: start PortRealEventStreamImpl, begin formulating how it will work
17-Dec-81 16:46:46: basic code compiles.  Yet to do: 1) monitor entry procedures, 2) define an interface and export to it, 3) finalization code to arrange to call stop (and nill out the pointer) (or call destroy) 4) opposite direction streams.
21-Dec-81 13:24:26: now exports PortRealEventStream
21-Dec-81 13:42:11: add object monitor.  UGH, since monitor lock expression must be uniform over all entry procedures, and since I did not want to split into more than one file, I have had to combine the two type stream object records, leading to waste space.  UGH
21-Dec-81 14:36:32 made Howard upset about Cedar and its performance. Converted to RefAnyStream. PDR
21-Dec-81 15:39:32: renamed all things to be RefAnyFoo rather than RealEventFoo
21-Dec-81 16:52:59: modified ports to pass shut down signals (out of band) in both directions.
21-Dec-81 17:55:48: renamed module to RefAnyStreamImpl. Renamed XXX to CreateGeneratorDrivenStream, exported to RefAnyStream. Made PortRealEventStream go away. Put in calls on Error in place of ERROR. PDR
22-Dec-81 11:30:41: confusion on shutdown indication, removed continue field, and will use sink=NIL as signal for consumer requested termination.
22-Dec-81 12:24:01: rename generator to be producer, and add consumer driven stream (opposite sense from first version)
22-Dec-81 14:29:53: change nomenclature so as to distinguish producer/consumer, coforked/client, etc issues
�Ê
e˜Jš
œ™Jš
œ.™.Jš
œ2™2J˜š
Ïk	˜	Jšœ�œ˜Jšœ
�œ)˜;J˜—š	œ�œ�œ
�œ�œ˜^Jš�œ˜Jš�œ˜Jš
�˜J˜š
œ™Jš
œ/™/Jš
œ.™.J˜Jš
œ™Jš
œ™J˜Jš
œP™PJš
œT™TJ˜Jš
œM™MJš
œP™PJ˜——Jš
œ™Jšœ�œ
�œ �œ
˜8J˜Jš
œD™D˜Jš
œB™BJ˜J˜Jšœ�œ�œ
�œ
�œ�œ	�œ
�œ�œ�œ	�œ
�œÏc˜�
Jšœ�œ�œ
�œ
�œ�œ	�œ
�œ�œ�œ	�œ
�œž˜€
Jš
Ïn
œ�œ�	œ�œ	�œ
�œž˜UJ˜Jš
œÙ
™Ù
J˜Jšœ�œ�œ˜&J˜šœ�œ�	œ
�œ
˜(š
œ.™.J˜J˜J˜—š
œ)™)Jšœ�œ
˜Jšœ�œ
�œ
˜J˜—š
œ1™1J˜J˜J˜J˜——Jš
œ?™?J˜J˜Jšœ�œ�œß
˜žJ˜šŸœ�	œ�œ
�œ˜@Jšœ
�œ%˜+J˜—šŸœ�œ
�	œ�œ
�œ
�œ˜RJš	œ
�œ
�œ�œ�œ%˜BJ˜—š
Ÿœ�œ
�	œ#�œ
�œ˜QJš�œ
�œ
�œ�œ˜Jšœ�œ˜"Jš�œ
�œ
�œ�œ
�œ	�œ�œ
�œ�œ
�œ˜nJš�œ
˜J˜—š
Ÿœ�œ
�	œ#�œ
�œ˜UJš	œ
�œ
�œ�œ�œ%˜BJ˜—šŸœ�œ
�	œ˜CJš	œ
�œ
�œ�œ�œ%˜BJ˜—šŸœ�œ
�	œ˜CJš�œ
�œ
�œ�œ˜Jšœ�œ˜"Jšœ	�œ
˜
Jš�œ
˜J˜—š
Ÿœ�œ
�	œ�œ
�œ˜TJš�œ
�œ
�œ�œ˜Jšœ�œ˜"Jš�œ�œ˜Jš�œ
˜J˜J˜J˜—Jš
œ5™5J˜J˜Jšœ�œ�œÒ
˜�J˜šŸœ�œ
�	œ�œ
�œ
�œ˜PJš�œ
�œ
�œ�œ˜Jšœ�œ˜"Jš�œ�œ
�œ
�œ˜1Jš�œ
�œ�œ�œ
˜:Jšœ!�œ�œ˜,š�œ˜Jš�œ�œ�œ	˜*š�œ	�œ�œ˜Jšœ�œ
˜Jš�œ˜——Jš�œ
˜J˜—š
Ÿœ�œ
�	œ#�œ
�œ˜OJš	œ
�œ
�œ�œ�œ%˜BJ˜—š
Ÿœ�œ
�	œ#�œ
�œ˜SJš
œ
�œ
�œ�œ�œ&ž˜UJ˜—šŸœ�œ
�	œ˜AJš	œ
�œ
�œ�œ�œ%˜BJ˜—š
Ÿœ�œ
�	œ�œ
�œ˜RJš�œ
�œ
�œ�œ˜Jšœ�œ˜"Jš�œ
�œ
�œ
�œ˜"Jš
�œ�œ
�œ
�œ
�œ˜$Jšœ!�œ�œ˜,Jš�œ
�œ
˜Jš�œ
˜J˜—šŸœ�œ
�	œ˜AJš�œ
�œ
�œ�œ˜Jšœ�œ˜"š�œ�˜š�œ	�œ�œ˜Jšœ�œ˜——Jš�œ
˜J˜J˜J˜J˜—Jš
œ™J˜šŸœ�	œ�œ
�œ
�œ�œ
�œ�œ3˜—
Jš
�˜J˜%Jšœ�œ�œ�œ�œ�œ	˜FJ˜ Jš�œ
�œ�œ�œ�œ
�œ
˜5Jšœ
�œF˜VJ˜Jš�œ
˜J˜—šŸœ�œ
�œ�œ�œ
�œž%œ	�œ
�œ
˜—
Jš�œž"œ˜@Jš
�˜J˜J˜J˜J˜"J˜6Jš	œ�œ$�œ�œ�œ˜=Jšœ�œD˜VJš�œ˜Jš�œ
˜J˜J˜J˜——Jš
œ<™<J˜˜Jšœ�œ�œÓ
˜�J˜J˜šŸœ�œ
�	œ�œ
�œ
�œ˜PJš	œ
�œ
�œ�œ�œ%˜BJ˜—š
Ÿœ�œ
�	œ#�œ
�œ˜OJš�œ
�œ
�œ�œ˜Jšœ�œ˜"Jš�œ�œ
�œ�œ
�œ�œ�œ�œ�œ
�œ˜pJš�œ
˜J˜—š
Ÿœ�œ
�	œ#�œ
�œ˜SJš	œ
�œ
�œ�œ�œ%˜BJ˜—šŸœ�œ
�	œ˜AJš	œ
�œ
�œ�œ�œ%˜BJ˜—šŸœ�œ
�	œ˜AJš�œ
�œ
�œ�œ˜Jšœ�œ˜"š�œ�˜š�œ	�œ�œ˜Jšœ�œ˜——Jš�œ
˜J˜—š
Ÿœ�œ
�	œ�œ
�œ˜RJš�œ
�œ
�œ�œ˜Jšœ�œ˜"Jš�œ
�œ˜Jš�œ
˜J˜J˜J˜—Jš
œ8™8J˜J˜Jšœ�œ�œÞ
˜�J˜šŸœ�œ
�	œ�œ
�œ
�œ˜RJš�œ
�œ
�œ�œ˜Jšœ�œ˜"Jš�œ
�œ
�œ
�œ˜/Jš�œ
�œ�œ�œ
˜:Jšœ�œ�œ˜*š�œ˜Jš�œ�œ�œ	˜*Jš�œ�œ�œ˜/—Jš�œ
˜J˜—š
Ÿœ�œ
�	œ#�œ
�œ˜QJš	œ
�œ
�œ�œ�œ%˜BJ˜—š
Ÿœ�œ
�	œ#�œ
�œ˜UJš
œ
�œ
�œ�œ�œ&ž˜UJ˜—šŸœ�œ
�	œ˜CJš	œ
�œ
�œ�œ�œ%˜BJ˜—š
Ÿœ�œ
�	œ�œ
�œ˜TJš�œ
�œ
�œ�œ˜Jšœ�œ˜"Jš�œ
�œ
�œ
�œ˜"Jš
�œ
�œ
�œ
�œ
�œ˜"Jšœ!�œ�œ˜,Jš�œ
�œ
�œ
�œ
˜%Jš�œ
�œ
˜Jš�œ
˜J˜—šŸœ�œ
�	œ˜CJš�œ
�œ
�œ�œ˜Jšœ�œ˜"Jšœ	�œ
˜
Jš�œ
˜J˜J˜Jš
œ™J˜—šŸœ�	œ�œ
�œ
�œ�œ
�œ�œ3˜—
Jš
�˜J˜%Jšœ�œ�œ�œ�œ�œ	˜FJ˜ Jš�œ
�œ�œ�œ�œ
�œ
˜5Jšœ
�œF˜VJ˜Jš�œ
˜J˜—šŸœ�œ
�œ�œ�œ
�œž%œ	�œ
�œ
˜™
Jš�œž"œ˜@Jš
�˜J˜J˜J˜J˜"J˜6Jš	œ�œ$�œ�œ�œ˜=Jšœ�œD˜VJš�œ˜Jš�œ
˜J˜J˜J˜J˜—Jš�œ
˜J˜Jš
œU™UJš
ϗ
™ó
Jš
œ3™3Jš
œˆ™ˆJš
œd™dJš
œO™OJš
œ^™^Jš
œÌ
™Ì
Jš
œ‘
™‘
Jš
œw™wJš
œk™kJ˜——…—#;Ù