-- File: PtBSP.mesa,  Last Edit:
  -- MAS  April 21, 1980  6:23 PM
  -- HGM  November 12, 1979  5:37 PM

DIRECTORY
  IODefs: FROM "IODefs" USING [WriteChar, WriteLine, WriteOctal, WriteString],
  MiscDefs: FROM "MiscDefs" USING [CallDebugger],
  ProcessDefs: FROM "ProcessDefs" USING [Detach],
  Stream: FROM "Stream" USING [
    GetBlock, PutBlock, SendNow,
    Block, Handle, TimeOut, CompletionCode, SubSequenceType],
  StatsDefs: FROM "StatsDefs" USING [StatBump],
  PupStream: FROM "PupStream" USING [
    CreatePupByteStreamListener, RejectThisRequest, DestroyPupListener,
    PupAddress, PupListener, StreamClosing,
    PupByteStreamCreate, PupByteStreamMake, SecondsToTocks],
  PtDefs: FROM "PtDefs" USING [
    blockCycle, blockSize, CursorBits, DataBuffer,
    PtInterface, PtLookerReady, UnListen, Random, Done, WaitUntilDone,
    recvIndex, SeeStorms, sendIndex, slow];

PtBSP: PROGRAM [pt: PtDefs.PtInterface]
  IMPORTS IODefs, MiscDefs, ProcessDefs, Stream, StatsDefs, PupStream, PtDefs
  EXPORTS PtDefs = BEGIN
OPEN pt, IODefs, StatsDefs, PupStream, PtDefs;

listener: PupListener _ NIL;
secondPushBS: Stream.Handle _ NIL;
globalPushBS: Stream.Handle _ NIL;
globalPullBS: Stream.Handle _ NIL;


Header: PROCEDURE [s, t: STRING, where: PupAddress] =
  BEGIN
  WriteString[s];
  SELECT mode FROM
    byte => WriteString[" bytes"];
    block => WriteString[" blocks"];
    ENDCASE => ERROR;
  IF where.host=myHost AND where.net=myNet
  THEN WriteString[" locally"]
  ELSE BEGIN WriteString[" "]; WriteString[t];
     IF where.net#myNet
      THEN BEGIN WriteString[" net "]; WriteOctal[where.net] END;
    WriteString[" host "]; WriteOctal[where.host]; END;
  WriteLine["."];
  SeeStorms[];
  END;

Push: PROCEDURE =
  BEGIN
  pushBS: Stream.Handle _ NIL;
  buffer: DataBuffer;
  block: Stream.Block _ [@buffer,0,blockSize];
  pushI, len, k: CARDINAL;
  r: CARDINAL _ 0;
  BEGIN  -- for EXITS
  pushBS _ PupByteStreamCreate[pushHim,SecondsToTocks[1] !
    StreamClosing =>
      BEGIN
      WriteString["Connection failed: "];
      IF text#NIL THEN
        BEGIN
        WriteChar['(];
        WriteString[text];
        WriteString[") "];
        END;
      WriteLine[whyText[why]];
      GOTO Failed;
      END];
  globalPushBS _ pushBS;
  WriteChar['O];
  SELECT length FROM
    short, cyclic => len _ 0;  -- start at short
    long, ignore => len _ blockSize;
    random => NULL;
    ENDCASE => ERROR;
  SELECT data FROM
    alternating => FOR k IN [0..blockSize) DO buffer[k] _ 125B; ENDLOOP;
    ones => FOR k IN [0..blockSize) DO buffer[k] _ 377B; ENDLOOP;
    zeros => FOR k IN [0..blockSize) DO buffer[k] _ 0; ENDLOOP;
    random => NULL;
    cyclic => NULL;
    ignore => NULL;
    ENDCASE => ERROR;
  UNTIL stopFlag DO
    ENABLE StreamClosing =>
      BEGIN
      WriteString["Push Stream closed: "];
      IF text#NIL THEN
        BEGIN
        WriteChar['(];
        WriteString[text];
        WriteString[") "];
        END;
      WriteLine[whyText[why]];
      GOTO Closed;
      END;
    SELECT mode FROM
      block =>
        BEGIN
        FOR pushI IN [1..blockCycle) UNTIL stopFlag DO
          SELECT length FROM
            short, long, ignore => NULL;
            cyclic => len _ pushI;
            random => len _ ((r_Random[r]) MOD blockSize+1);
            ENDCASE => ERROR;
          SELECT data FROM
            ignore => NULL;
            cyclic => FOR k IN [0..len) DO buffer[k] _ k+pushI; ENDLOOP;
            ones => NULL;
            zeros => NULL;
            alternating => NULL;
            random => FOR k IN [0..len) DO buffer[k] _ (r_Random[r]); ENDLOOP;
            ENDCASE => ERROR;
          block.stopIndexPlusOne _ len;
          Stream.PutBlock[pushBS,block,FALSE];
          IF ~doStats THEN StatBump[statDataBytesSent,blockSize];
          IF slow THEN CursorBits[sendIndex] _ CursorBits[sendIndex]+1;
          ENDLOOP;
        END;
      ENDCASE => ERROR;
    IF info THEN WriteChar['S];
    Stream.SendNow[pushBS];
    IF ~slow THEN CursorBits[sendIndex] _ CursorBits[sendIndex]+1;
    ENDLOOP;
  EXITS Closed, Failed => NULL;
  END;
  IF pushBS#NIL THEN pushBS.delete[pushBS];
  IF secondPushBS=pushBS THEN secondPushBS_NIL;
  IF globalPushBS=pushBS THEN globalPushBS_NIL;
  Done[];
  END;

Pull: PROCEDURE [pullBS: Stream.Handle] =
  BEGIN
  listener: BOOLEAN _ pullBS#NIL;
  buffer: DataBuffer;
  block: Stream.Block _ [@buffer,0,blockSize];
  pullI, len, k: CARDINAL;
  r: CARDINAL _ 0;
  bytes: CARDINAL;
  why: Stream.CompletionCode;
  sst: Stream.SubSequenceType;
  IF pullBS=NIL THEN
    BEGIN
    pullBS _ PupByteStreamMake[
        pullMe.socket,
        pullHim,
        SecondsToTocks[1],
        wait,
        [0,0]];
    globalPullBS _ pullBS;
    END;
  SELECT length FROM
    short, cyclic => len _ 0;  -- start at short
    long, ignore => len _ blockSize;
    random => NULL;
    ENDCASE => ERROR;
  UNTIL stopFlag DO
    ENABLE StreamClosing =>
      BEGIN
      WriteString["Pull Stream closed: "];
      IF text#NIL THEN
        BEGIN
        WriteChar['(];
        WriteString[text];
        WriteString[") "];
        END;
      WriteLine[whyText[why]];
      GOTO Closed;
      END;
    SELECT mode FROM
      block =>
        BEGIN
        FOR pullI IN [1..blockCycle) UNTIL stopFlag DO
          SELECT length FROM
            short, long, ignore => NULL;
            cyclic => len _ pullI;
            random => len _ ((r_Random[r]) MOD blockSize+1);
            ENDCASE => ERROR;
          SELECT data FROM
            ignore => NULL;
            cyclic => NULL;
            ones, zeros, alternating, random =>
                FOR k IN [0..len) DO buffer[k] _ 123; ENDLOOP;
            ENDCASE => ERROR;
          block.stopIndexPlusOne _ len;
          [bytes,why,sst] _ Stream.GetBlock[pullBS,block !
              Stream.TimeOut =>
                BEGIN
                IF info THEN WriteChar['?];
                RESUME;
                END ];
          SELECT why FROM
            normal => NULL;
            endRecord => BEGIN MiscDefs.CallDebugger["endRecord -- ???"]; END;
            sstChange => BEGIN MiscDefs.CallDebugger["sstChange -- ???"]; END;
            ENDCASE => ERROR;
          IF bytes#len THEN BEGIN MiscDefs.CallDebugger["wrong length -- ???"]; END;
          SELECT data FROM
            ignore => NULL;
            cyclic => FOR k IN [0..len) DO
              IF buffer[k]#((k+pullI) MOD 400B) THEN ERROR; ENDLOOP;
            ones => FOR k IN [0..len) DO
              IF buffer[k]#377B THEN ERROR; ENDLOOP;
            zeros => FOR k IN [0..len) DO
              IF buffer[k]#0 THEN ERROR; ENDLOOP;
            alternating => FOR k IN [0..len) DO
              IF buffer[k]#125B THEN ERROR; ENDLOOP;
            random => FOR k IN [0..len) DO
              IF buffer[k]#((r_Random[r]) MOD 400B) THEN ERROR; ENDLOOP;
            ENDCASE => ERROR;
          IF slow THEN CursorBits[recvIndex] _ CursorBits[recvIndex]+1;
          IF ~doStats THEN StatBump[statDataBytesReceived,bytes];
          -- check buffer now
          ENDLOOP;
        END;
      ENDCASE => ERROR;
    IF ~slow THEN CursorBits[recvIndex] _ CursorBits[recvIndex]+1;
    IF info THEN WriteChar['R];
    REPEAT Closed => NULL;
    ENDLOOP;
  pullBS.delete[pullBS];
  IF globalPullBS=pullBS THEN globalPullBS _ NIL;
  IF ~listener THEN Done[];
  END;

Hello: PROCEDURE [bs: Stream.Handle, remote: PupAddress] =
  BEGIN
  IF info THEN Header["receiving","from",remote];
  ProcessDefs.Detach[FORK Pull[bs]];
  END;

PtBSPListen: PUBLIC PROCEDURE =
  BEGIN
  listener _ CreatePupByteStreamListener[socNum,Hello,SecondsToTocks[1]];
  END;

PtBSPPushPull: PUBLIC PROCEDURE =
  BEGIN
  UnListen[];
  IF pullHim.socket=[0,0] THEN BEGIN
    WriteLine["Need a socket number."]; RETURN; END;
  Header["Exchanging","with",pushHim];
  PtLookerReady[];
  ProcessDefs.Detach[FORK Push[]];
  ProcessDefs.Detach[FORK Pull[NIL]];
  WaitUntilDone[2];
  END;

PtBSPRecv: PUBLIC PROCEDURE =
  BEGIN
  UnListen[];
  IF pullHim.socket=[0,0] THEN BEGIN
    WriteLine["Need a socket number."]; RETURN; END;
  Header["Receiving","from",pullHim];
  PtLookerReady[];
  ProcessDefs.Detach[FORK Pull[NIL]];
  WaitUntilDone[1];
  END;

PtBSPSend: PUBLIC PROCEDURE =
  BEGIN
  Header["Sending","to",pushHim];
  PtLookerReady[];
  ProcessDefs.Detach[FORK Push[]];
  WaitUntilDone[1];
  END;

PtBSPDoubleSend: PUBLIC PROCEDURE =
  BEGIN
  Header["Sending double","to",pushHim];
  PtLookerReady[];
  ProcessDefs.Detach[FORK Push[]];
  ProcessDefs.Detach[FORK Push[]];
  WaitUntilDone[2];
  END;


PtBSPUnListen: PUBLIC PROCEDURE =
  BEGIN
  IF listener=NIL THEN RETURN[];
  DestroyPupListener[listener]; listener_NIL;
  WriteLine["ByteStream Unlisten."];
  END;

rejecter: PupListener _ NIL;

RejectEverything: PROCEDURE  [PupAddress] =
  BEGIN
  ERROR RejectThisRequest["Reject this turkey"];
  END;


PtRejListen: PUBLIC PROCEDURE =
  BEGIN
  rejecter _ CreatePupByteStreamListener[
      socNum,Hello,SecondsToTocks[1],RejectEverything];
  END;

PtRejUnListen: PUBLIC PROCEDURE =
  BEGIN
  IF rejecter=NIL THEN RETURN[];
  DestroyPupListener[rejecter]; rejecter_NIL;
  WriteLine["Rejector Unlisten."];
  END;



-- here is what gets STARTed
END.(0,65535)(1,3528)(2,4057)(9,12347)\107t2 1t0 755t7 4t0 7t7 11t0 110t7 2t2 2t0 274t2 94t0 211t2 4t0 151t2 10t0 10t2 11t0 36t2 2t0 371t3 6t0 11t2 3t0 25t2 17t0 214t2 6t0 60t2 6t0 61t2 6t0 100t2 2t0 18t4 4t0 267t4 4t0 17t4 6t0 23t4 8t0 500t2 6t0 60t4 50t0 37t4 10t0 66t2 3t0 4t2 55t4 10t0 22t4 6t3 18t4 2t0 30t4 2t0 30t2 3t0 5t2 55t4 2t0 20t3 14t0 9t4 2t0 248t2 10t0 10t2 11t0 36t2 2t0 519t3 6t0 6t3 6t0 164t3 6t0 25t3 6t0 13t3 6t0 26t3 6t0 77t4 1t0 364t2 6t0 50t4 40t0 51t5 78t0 27t4 1t5 44t0 264t4 3t0 577t2 6t0 90t2 3t0 4t2 17t0 9t2 15t0 9t2 3t0 78t4 20t0 10t4 17t0 5t3 24t0 4t2 3t0 5t2 17t0 9t2 15t0 9t2 3t0 33t4 1t0 12t3 5t0 1571t10 32t0 17t10 31t0