Slipstream.Lib.dog

//////////////////  Classes for Slipstream Based Applications
// Slipstream.Lib.dog

//LinuxTestBuild: Platform='Linux' CPU='amd64' Lang='CPP' testMode='makeTests';
LinuxBuild:     Platform='Linux' CPU='amd64' Lang='CPP' LangVersion='GNU' optimize='speed';

FileName = "slipstreamCore"
Version = "0.1"
CopyrightMesg = "Copyright (c) 2020-<YEAR> Bruce Long"
Authors = "Bruce Long"
Description = "The Slipstream Engine"
ProgramOrLibrary = "program"

featuresNeeded = [Proteus, Logger, Unicode, BigNumbers, Threads, IOBuffer, Quic] // P2P_Manager]

LicenseText = `This file is part of the "Slipstream Development Suite" All Rights Reserved.`

runCode=`
    log("Slipstream CORE started")
    CommandLineManager.processCmdLine(joinCmdStrings(argc, argv), false);
    CommandLineManager.defineOption("Slip", "topFile", "-f", "--file", "The top-level file to load.", "")
    me string: topFile <- CommandLineManager.getOption("Slip", "topFile")
    if(topFile==""){topFile <- "world.pr"}
    CORE.run(topFile)
`
do ManageCmdLine(cmd)

struct GLOBAL{
    me SlipstreamCore: CORE
}
struct Repository{ // Load, track, save, pull, push, pullReq, etc.
    me string: repoName
    me string: repoLocator
    our infon: repoInfon
    our InfonStream: iStream
    our infon: initAndLoad(their ConnectionManager: connectionMgr, me string: name, me string:locator) <- {
        log("Starting repository:"+name+" at "+locator)
        Allocate(repoInfon)
        repoName    <- name
        repoLocator <- locator
        iStream <- connectionMgr.startNewConnection(repoInfon, repoLocator)
        log("Started repository:"+name+" at "+repoLocator)
        return(repoInfon)
    }
    our InfonStream: appendAnInfonAndLoad(their ConnectionManager: connectionMgr, me string:locator) <- {
  ///      if(repoInfon.value.tailUnfinished == false){logFatalError("Repo infon must be unfinished with '...'.\n")}
        our infon:: newInf
        // Ensure the new infon won't get processed by other threads then append it to repoInfon
        our InfonStream: newiStream <- connectionMgr.startNewConnection(newInf, locator)
        return(newiStream)
    }
}

#include Protocols.dog

model InfonConnector{
    our SStreamSpec: streamSpec
    me strBuf: inputBuf
    me strBuf: outputBuf
    me int:    statusCode
    me string: statusMesg
    me bool:   inputEOF
    their InfonStream: iStreamParent
    void: init(our SStreamSpec: SP) <- {}
    void: pumpInputBuff()        <- {}
    me string: readInputStream() <- {return("")}
    bool: pumpOutputBuf(me string: mesg)  // send to outputstream: outputBuf.putStr(mesg);
    void: submitProteusResult(me int64: pos, me string: mesg)  <- {pumpOutputBuf("RESULT:"+toString(pos)+":"+mesg+"\n")}
    void: handleSystemError  (me string: mesg)  <- {pumpOutputBuf("SYSERR:"+mesg+"\n")}
    void: handleSyntaxError  (me string: mesg)  <- {pumpOutputBuf("SYNERR:"+mesg+"\n")}
    void: handleProteusError (me string: mesg)  <- {pumpOutputBuf("NRMERR:"+mesg+"\n")}
}
struct InfonConnector{}

struct FileConnector: inherits=InfonConnector {
    me FileStream: dataFile
    void: init(our SStreamSpec: SP) <- {
        streamSpec <- SP
        me string: filePath <- getAssetsDir() +"/"+streamSpec.contentArgs
        dataFile.open(filePath)
        if (dataFile.failure()){logFatalError("Failed to open file: " + filePath + "\n")}
        inputEOF<-false
        log("Initialized:"+filePath)
    }
    void: pumpInputBuff() <- {
        while(iStreamParent.pumpParserInputBuf()){}
        inputBuf.close()
    }
    me string: readInputStream()  <- {
        me string: data
        if(!dataFile.EOF()){
            data <- dataFile.getLine()+"\n"
       // log("READ:"+to_string(data.size())+" chars from:"+streamSpec.spec+"\t"+data)
        } else{inputEOF<-true; log("INPUT_EOF:"+streamSpec.spec)}
        return(data)
    }
    bool: pumpOutputBuf(me string: mesg) <- {print(mesg); return(false)}
}
struct ClipConnector: inherits=InfonConnector {
    void: init(our SStreamSpec: SP) <- {
        streamSpec <- SP
        inputEOF<-false
        log("Initialized CLIP connection")
    }
    void: pumpInputBuff() <- {
        while(iStreamParent.pumpParserInputBuf()){}
        inputBuf.close()
    }
    me string: readInputStream()  <- {
        me string: data
        input(data)
        if(data=="quit"){inputEOF<-true; return("")}
        return(data+"\n")
    }
    bool: pumpOutputBuf(me string: mesg) <- {print(mesg); return(false)}
}
struct DrctConnector: inherits=InfonConnector {     // Direct / Local connection
    our TerminalApp: termToConnect  //TODO: This lib shouldn't depend on TerminalApp; use a proxie class
    void: init(our SStreamSpec: SP) <- {
        streamSpec <- SP
    }
    bool: pumpOutputBuf(me string: mesg) <- {
        termToConnect.processResult(mesg)
        slipView.requestRedraw();
        return(false)
    }
}
//~ struct UserConnector: inherits=InfonConnector {}
//~ struct IPFSConnector: inherits=InfonConnector {}
//~ struct IPNSConnector: inherits=InfonConnector {}
//~ struct RepoConnector: inherits=InfonConnector {}     // Connector to a Repository instance.
//~ struct GUI_Connector: inherits=InfonConnector {}     // Kbd, mouse/touchScn, screen, clipboard, etc.
//~ struct MediaConnector:inherits=InfonConnector {}     // Microphones, audioOut, videoSource
struct AgentConnector:inherits=InfonConnector {      // Connect to other agents
    void: init(our SStreamSpec: SP) <- {
        streamSpec <- SP
        their QuicClient_Connection: qConn  <- CORE.quicClient.getClientConnection(SP.ipAddr, SP.portNum)
        if(qConn==NULL){log("ERROR setting up QuicClient_Connection."); return()}

        our FetchFile:: fetchFile
        fetchFile.setArgs(SP.contentArgs)
        fetchFile.source <- self
        qConn.enQSession(fetchFile)
        CORE.quicClient.start()
    }
    bool: pumpOutputBuf(me string: mesg) <- {print(mesg); return(false)}
}
//~ struct SensorConnector  : inherits=InfonConnector {}
//~ struct RobotConnector   : inherits=InfonConnector {}
//~ struct InternetConnector: inherits=InfonConnector {}
//~ struct PhoneConnector   : inherits=InfonConnector {} // Services offered by phones


//////////////////////////////////////////////////////////////////////

struct ThreadedInfonParser: inherits=Threads{
    me string: name
    our ProteusParser: parser
    our infon: result
    our InfonConnector: sourceConnector
    me bool: parseCompleted
    me Mutex: chkParseDone
    me SyncLock: parseDoneLock

    void: waitForParseCompletion()<-{
        me MutexMngr: MtxMgr{chkParseDone}
        while(!parseCompleted){
            parseDoneLock.wait(MtxMgr)
        }
    }

    void: initAndStart(our ProteusParser: pParser, their strBuf: streamToParse, our infon: topItem, me string:Name) <- {
        parser <- pParser
        result <- topItem
        name <- Name+"_P"
        parseCompleted <- false
        parser.initParseFromStream(streamToParse)
        start(name)
    }

    void: run() <- {
        log("OPENING PARSE_THREAD:" + streamName())
        parser.doParse(result)
        if(parser.doesParseHaveError()){
            handleSyntaxError(parser.errorMesg)
        }
        log("CLOSING PARSE_THREAD:" + streamName())

        protect(chkParseDone){
            parseCompleted <- true;
            parseDoneLock.notifyOne();
        }
    }
    me string: streamName() <- {return(sourceConnector.streamSpec.spec)}
    void: handleSyntaxError(me string: mesg) <- {
        log("Syntax Error:"+mesg)
        sourceConnector.handleSyntaxError(streamName() + mesg)
    }
}
struct InfonStream{ // InfonConnector -> Parse -> Extract -> Normalize
    me string: name
    our InfonConnector: source
    me uint64: originID
    their Agent: parentAgent
    our ProteusParser: parser
    our infon: toFill
    me ThreadedInfonParser: parserThread
    me ThreadedNormalizer: normerThread
    bool: isActive  // True is a thread is still operating.

    bool: pumpParserInputBuf() <- {
        me string: data <- source.readInputStream()
        if(!source.inputEOF and data.size()>0){
            source.inputBuf.putStr(data)
        }
        return(!source.inputEOF)
    }
    bool: pushParserInputBuf(me string: data) <- {
        if(!source.inputEOF and data.size()>0){
            source.inputBuf.putStr(data)
        }
        return(!source.inputEOF)
    }

    void: init(our InfonConnector: connector, their Agent: agent, our ProteusParser: pParser, our infon: infToFill, me uint64: OriginID) <- {
        source   <- connector
        originID <- OriginID
        parentAgent <- agent
        parser <- pParser
        toFill <- infToFill
        source.iStreamParent <- self
        if(toFill==NULL){Allocate(toFill)}
        toFill.originID <- originID
        parserThread.sourceConnector <- source
    }
    void: start(me string:name) <- {
        toFill.stillParsing <- true
        normerThread.init(parentAgent, toFill)
        normerThread.start(name+"_NRM")
        parserThread.initAndStart(parser, source.inputBuf, toFill, name+"_PnE")
        source.pumpInputBuff()
        //waitForCompletion(); log("Result:"+toString(toFill))
    }
    void: waitForCompletion() <-{
        parserThread.waitForExit()
        normerThread.waitForExit()
    }
    void: requestFinish() <- {} // For long-running streams, request to end the stream.
}

struct ConnectionManager{ // Manage connections to a Proteus Agent
    their Agent: agent
    me Repository: device
    me Repository: user
    me Repository: world
    me int: numFilesLoaded <- 0
    me Mutex: connMgrMtx

    me uint64: lastOriginID
    me Map<uint64, our InfonStream>: infonSourceRecords // Maps infon::originID to the stream it came from.

    void: init(their Agent: agnt) <- {
        agent <- agnt
        lastOriginID <- 0
    }
    our InfonStream: startNewConnection(our infon: toFill, me string: specString) <- {
        me string: parentPath <- ""
        if(specString[0]=="/" and infonSourceRecords.containsKey(toFill.originID)){
            parentPath <- infonSourceRecords[toFill.originID].source.streamSpec.spec
            // Check for issues and remove slug from parentPath
            me int: lastSlashPos <- findLastSubString(parentPath, "/")
            if(lastSlashPos>=0){
                parentPath <- parentPath.subStr(0, lastSlashPos+1)
            }
        }
        log("Starting Connection:"+specString)
        our InfonStream:: iStream
        protect(connMgrMtx){
            lastOriginID <+- 1
            //TODO: Shouldn't need the next line as a proxy. Fix CodeDog bug by checking modelArgs in codeNewVarStr() for {} constructor calls.
            their SlipstreamCore: tmpCore <- CORE
            our ProteusParser:: parser{tmpCore, true, lastOriginID}
            our InfonConnector: connector <- makeInfonConnector(specString, parentPath)
            iStream.init(connector, agent, parser, toFill, lastOriginID)
            infonSourceRecords[lastOriginID] <- iStream
            iStream.start("IS"+toString(numFilesLoaded))
        }
        return(iStream)
    }

    our InfonConnector: makeInfonConnector(me string: specString, me string: parentPath)<-{ // connector factory
// if there is no schema: and no '//' concat specString and parentPath.
//TODO: gracefully handle errors
        if(specString[0]=="/"){specString <- specString.subStr(1)}
        numFilesLoaded <+- 1
        print("LOADING:("+toString(numFilesLoaded)+") "+parentPath+ specString+"\n")
        our SStreamSpec:: SP; SP.setSpec(parentPath+specString)

        me string: protocolTag <- SP.protocolTag

        switch(protocolTag){
            case "file":{our FileConnector:: retval;  retval.init(SP); return(retval)}
            case "clip":{our ClipConnector:: retval;  retval.init(SP); return(retval)}
            //~ case "repo":{our RepoConnector:: retval;  retval.init(SP); return(retval)}
            //~ case "user":{our UserConnector:: retval;  retval.init(SP); return(retval)}
            case "drct":{our DrctConnector:: retval;  retval.init(SP); return(retval)}
            //~ case "ipfs":{our IPFSConnector:: retval;  retval.init(SP); return(retval)}
            //~ case "ipns":{our IPNSConnector:: retval;  retval.init(SP); return(retval)}
            //~ case "quic":{our AgentConnector:: retval; retval.init(SP); return(retval)}
            case "agnt":{our AgentConnector:: retval; retval.init(SP); return(retval)}
        }
        log("WARNING, No connection was made for: "+specString)
        return(NULL)
    }
}

struct SlipstreamCore: inherits=ProteusCore{ // Create and manage a slipstream node
    me Agent: agent
    me string: masterFile
    me ConnectionManager: connectionMgr
    me string: statusMsg
    me bool: idling

    me bool: normalizeFromStreamSpec(our POV: infToFill, me string: streamSpec) <- {
        me bool: success <- connectionMgr.startNewConnection(infToFill.pItem, streamSpec)==NULL
        return(success)
    }
    me string: getOriginIDString(me uint64: originID) <- {
        return(connectionMgr.infonSourceRecords[originID].source.streamSpec.spec)
    }
    me QuicClient_Engine: quicClient

    void: submitProteusResult(me uint64: originID, me uint64: pos, me string: mesg)  <- {connectionMgr.infonSourceRecords[originID].source.submitProteusResult(pos, mesg)}
    void: handleSystemError  (me uint64: originID, me string: mesg)  <- {connectionMgr.infonSourceRecords[originID].source.handleSystemError(mesg)}
    void: handleSyntaxError  (me uint64: originID, me string: mesg)  <- {connectionMgr.infonSourceRecords[originID].source.handleSyntaxError(mesg)}
    void: handleProteusError (me uint64: originID, me string: mesg)  <- {connectionMgr.infonSourceRecords[originID].source.handleProteusError(mesg)}

    void: initCoreServices()<-{
        // Init P2P connections and DHTs
        // Etc.
    me string: ipAddr <- "::1"
    me string: portNum<- slipnode_default_port
        quicClient.init("CFile", false, ipAddr, stoi(portNum))
    }
    void: init(me string: topFile) <- {
        streamingNormMode <- true
        initHardFuncNames()
        statusMsg <- ""  // Empty == OK
        masterFile <- "file://"+topFile
        initCoreServices()
        agent.init(this)
        idling <- true
        connectionMgr.init(agent)
    }
    void: start() <- {
      //agent.device <- connectionMgr.device.initAndLoad(connectionMgr, "device", masterFile)
      //agent.world  <- connectionMgr.world.initAndLoad(connectionMgr,  "world","quic://[::1]:2006/public/PublicKB.pr")
        agent.world  <- connectionMgr.world.initAndLoad(connectionMgr,  "world","file:///public/PublicKB.pr")
        sleep(1000)
        agent.user   <- connectionMgr.user.initAndLoad(connectionMgr,   "user", "file:///userKB.pr")
       // sleep(1000)
      //agent.world  <- connectionMgr.world.initAndLoad(connectionMgr,  "world","clip://localhost/bruce")
    }
    void: idle()<-{
        me int: count<-0
        idling <- true
        while(idling){
            // process events if needed
            print(".")
            sleep(10)
count <+- 1
if(count>250){
    print("\nnumFilesLoaded:", connectionMgr.numFilesLoaded,"\n")
    stopIdling()
}
        }
    }
    void: stopIdling()<-{idling <- false}
    void: shutdown()<-{} // TODO: Ask all streams to quit then wait for them. Dealloc anything

    void: run(me string: topFile) <- {
        // This is an example or for testing. Normally use a GUI's event loop.
        init(topFile)
        start()
        idle()
        shutdown()
    }
}