<<>> <> <> <> <> DIRECTORY Basics, XCommunications, UnixSocket, UnixSysCalls, UXStrings, IO, Rope; XCommunicationsTCPPCedarNoYP: CEDAR PROGRAM IMPORTS Basics, IO, Rope, UnixSysCalls, XCommunications ~ BEGIN STREAM: TYPE = IO.STREAM; <> <> <> <> <> <<--C-- HostEnt: TYPE = MACHINE DEPENDENT RECORD [ >> <<--see chapter 9.2, page 233>> <> <> <> <> <> <<];>> --C-- PortNumber: TYPE = MACHINE DEPENDENT RECORD [val: Basics.HWORD]; --C-- InAddr: TYPE = PACKED ARRAY [0..4) OF BYTE; --internet address --C-- SockAddrIn: TYPE = WORD16 MACHINE DEPENDENT RECORD [ --see chapter 8.7, page 203 ... sinFamily: UnixSocket.ProtocolFamily, sinPort: PortNumber, sinAddr: InAddr, sinZero: PACKED ARRAY [0..8) OF BYTE ]; --C-- SockAddr: TYPE = WORD16 MACHINE DEPENDENT RECORD [ SELECT OVERLAID * FROM standard => [standard: UnixSocket.SockAddr], in => [in: SockAddrIn], ENDCASE ]; <> <> <<"gethostbyname">> <<};>> <

> <<};>> <<>> myStreamProcs: REF IO.StreamProcs Ź IO.CreateStreamProcs[ variety: inputOutput, class: $MyTCP, getChar: MyGetChar, endOf: MyEndOf, unsafeGetBlock: MyUnsafeGetBlock, putChar: MyPutChar, unsafePutBlock: MyUnsafePutBlock, flush: SendNow, close: MyClose ]; bufferSize: INT = 1024; MyDataRec: TYPE = MONITORED RECORD [ socket1: UnixSysCalls.FD, --writer... socket2: UnixSysCalls.FD, --reader... closed: BOOL Ź FALSE, bufCnt: INT Ź 0, server: SockAddrIn, <> outBuffer: PACKED ARRAY [0..bufferSize) OF CHAR ]; MyEndOf: PROC [self: STREAM] RETURNS [BOOL] = { RETURN [FALSE]; }; MyPutChar: PROC [self: STREAM, char: CHAR] = { d: REF MyDataRec = NARROW[self.streamData]; IF d.closed THEN ERROR IO.Error[StreamClosed, self]; IF d.bufCnt>=bufferSize THEN MyFlush[d]; d.outBuffer[d.bufCnt] Ź char; d.bufCnt Ź d.bufCnt+1; IF d.bufCnt>=bufferSize THEN MyFlush[d]; }; MyUnsafePutBlock: PROC [self: STREAM, block: Basics.UnsafeBlock] = TRUSTED { d: REF MyDataRec = NARROW[self.streamData]; IF d.closed THEN ERROR IO.Error[StreamClosed, self]; IF d.bufCnt>=bufferSize THEN MyFlush[d]; DO bCnt: INT Ź d.bufCnt; <<--copy that even competing processes will not cause MoveBytes to write outside buffer >> cnt: INT Ź MIN[bufferSize-bCnt, block.count]; MoveBytes[@d.outBuffer+bCnt, block.base+block.startIndex, cnt]; d.bufCnt Ź d.bufCnt + cnt; block.startIndex Ź block.startIndex + cnt; block.count Ź block.count - cnt; IF block.count>0 THEN MyFlush[d] ELSE EXIT ENDLOOP; IF d.bufCnt>=bufferSize THEN MyFlush[d]; }; MoveBytes: UNSAFE PROC [dest: POINTER, src: POINTER, len: CARDINAL] ~ UNCHECKED MACHINE CODE{ "XR_MoveBytesDisjoint" }; MyGetChar: PROC [self: STREAM] RETURNS [ch: CHAR] = TRUSTED { d: REF MyDataRec = NARROW[self.streamData]; n: INT; buff: PACKED ARRAY [0..BYTES[WORD]) OF CHAR; IF d.closed THEN ERROR IO.Error[StreamClosed, self]; n Ź UnixSysCalls.Read[d.socket2, LOOPHOLE[@buff], 1]; IF n = 0 THEN ERROR IO.EndOfStream[self]; IF n < 0 THEN ERROR; RETURN [buff[0]]; }; MyUnsafeGetBlock: UNSAFE PROC [self: STREAM, block: Basics.UnsafeBlock] RETURNS [nBytesRead: INT Ź 0] = { d: REF MyDataRec = NARROW[self.streamData]; IF d.closed THEN ERROR IO.Error[StreamClosed, self]; WHILE block.count>0 DO n: INT; TRUSTED { n Ź UnixSysCalls.Read[d.socket2, LOOPHOLE[block.base+block.startIndex+nBytesRead], block.count]; }; IF n<=0 THEN { IF n=0 THEN RETURN; ERROR; }; block.count Ź block.count-n; nBytesRead Ź nBytesRead+n; ENDLOOP }; MyClose: PROC [self: STREAM, abort: BOOL] = { d: REF MyDataRec = NARROW[self.streamData]; IF d.closed THEN RETURN; IF ~abort THEN SendNow[self]; d.closed Ź TRUE; [] Ź UnixSysCalls.Close[d.socket1]; [] Ź UnixSysCalls.Close[d.socket2]; }; MyFlush: PROC [d: REF MyDataRec] = TRUSTED { addr: POINTER Ź @d.outBuffer; cnt: INT Ź d.bufCnt; d.bufCnt Ź 0; WHILE cnt>0 DO n: INT Ź UnixSysCalls.Write[d.socket1, LOOPHOLE[addr], cnt]; IF n<=0 THEN ERROR; cnt Ź cnt-n; addr Ź addr+n; ENDLOOP }; SendNow: PROC [self: STREAM] = { d: REF MyDataRec = NARROW[self.streamData]; IF d.bufCnt>0 THEN MyFlush[d]; }; Create: PROC [base: Rope.ROPE, port: REF ANY] RETURNS [sd: XCommunications.StreamData] = TRUSTED { portNo: INT; d: REF MyDataRec Ź NEW[MyDataRec]; WITH port SELECT FROM ri: REF INT => portNo Ź ri; ENDCASE => {sd.errorMsg Ź "bad port type"; GOTO Oops}; <<--create socket>> d.socket1 Ź UnixSysCalls.Socket[inet, stream, unspec]; IF LOOPHOLE[d.socket1, INT]<0 THEN {sd.errorMsg Ź "socket not created"; GOTO Oops}; <<--bind socket to name>> d.server.sinFamily Ź inet; <> <> <> <> SELECT TRUE FROM Rope.Match["*clarissa*", base, FALSE] => d.server.sinAddr Ź [13, 1, 100, 140]; Rope.Match["*presto*", base, FALSE] => d.server.sinAddr Ź [13, 1, 100, 151]; Rope.Match["*lyrane*", base, FALSE] => d.server.sinAddr Ź [13, 1, 100, 213]; Rope.Match["*kimball*", base, FALSE] => d.server.sinAddr Ź [13, 1, 100, 206]; Rope.Match["*illona*", base, FALSE] => d.server.sinAddr Ź [13, 1, 101, 48]; ENDCASE => d.server.sinAddr Ź [13, 1, 100, 140]; --clarissa d.server.sinPort Ź [Basics.HFromInt16[portNo]]; <<--connect socket with remote socket>> IF UnixSysCalls.Connect[d.socket1, LOOPHOLE[@d.server], BYTES[SockAddrIn]] # success THEN { sd.errorMsg Ź "not connected"; GOTO Oops }; <<>> d.socket2 Ź UnixSysCalls.Dup[d.socket1]; IF LOOPHOLE[d.socket2, CARD32]=0 THEN { [] Ź UnixSysCalls.Close[d.socket1]; sd.errorMsg Ź "not connected: duplication of socket failed"; GOTO Oops }; <<>> <<--create stream>> sd.in Ź sd.out Ź IO.CreateStream[streamProcs: myStreamProcs, streamData: d]; sd.success Ź TRUE; sd.errorFromStream Ź ErrorFromStream; EXITS Oops => {} }; ErrorFromStream: PROC [s: IO.STREAM] RETURNS [reason: Rope.ROPE] = { reason Ź "failure"; }; XCommunications.RegisterCommunication[[create: Create, protocol: $TCP]]; END.