Include %BigData.ShardingManagerClass Patch.GlobalBinary Extends%Stream.TmpCharacter [ ClientDataType = BINARYSTREAM, OdbcType = LONGVARBINARY, System = 2 ]
{

Parameter COMPRESS = 1;
Method %OnNew(initval As%String = "") As%Status [ Private ]
{
    set i%%Location=$select(initval="":$$$streamGlobal,$$$UPPER(initval)="SHARD":"^IRIS.Stream.Shard",1:initval)
    RETURN$$$OK
}

Method %LoadData(id As%String) As%Status [ Private ]
{
    Set i%Id=id,i%NodeNo=0If i%%Location=""Set i%%Location=$$$streamGlobalIf id="" {
        Set i%Mode=$$$modeREADNODATA,i%AtEnd=1,i%IOSize=0,i%MaxNodeNo=0,i%Compress=..#COMPRESS
    } Else {
        If ('i%%Locked) && (i%%Concurrency) { Do ..%LockStream() }
        #; Turn on batch mode so reading stream does not kill the cacheSet batch=$zu(68,25,1)
        Set maxnode="",size=0,location=i%%LocationSet header=$get(@location@(id))
        If header["," {
            #; New header formatSet maxnode=$piece(header,","),size=+$piece(header,",",2),i%Compress=+$piece(header,",",3)
        } Else {
            #; Old header format has just maxnode in it
            #; For compression specific subclasses ..#COMPRESS=2 we always use compression even for old header formatSet i%Compress=$select(..#COMPRESS=2:2,1:-1)
            Set maxnode=$select(header="":+$order(@location@(id,""),-1),1:header)
            #; read size from "0" node, if present, if not calculate itIf maxnode'="",'$data(@location@(id,0),size) {
                Set size=0For i=1:1:maxnode Set size=size+$length($$$streamDecompressBuf(^(i)))
            }
        }
        Set i%IOSize=size,i%AtEnd='size,i%Mode=$select(maxnode="":$$$modeREADNODATA,1:$$$modeREADNOTCHANGED),i%MaxNodeNo=+maxnode
        Do$zu(68,25,batch)
    }
    Quit$$$OK
}

Method %SaveData(ByRef sid As%String) As%Status [ Private ]
{
    set tStatus = $$$OK, tLocked = 0
    #; save the stream;Set mode=i%ModeIf mode=$$$modeREADNODATA||(mode=$$$modeREADNOTCHANGED)||(mode=$$$modeWRITEJUSTSAVED) Quit$$$OKtry {
        #; Before changing anything we need to remember what the current state is in case of a rollbackKill i%rollbackIf$tlevelSet i%rollback("Buffer")=i%Buffer,i%rollback("Mode")=mode,i%rollback("MaxNodeNo")=i%MaxNodeNo
        #; Turn on batch mode so reading stream does not kill the cacheSet batch=$zu(68,25,1)
        If mode=$$$modeWRITE||(mode=$$$modeREADBUFFER) { Set buffer=i%Buffer } Else { Set buffer="" }
        If i%%Location=""Set i%%Location=$$$streamGlobalSet location=i%%Location
        #; kill original data and save size in "0" nodeSet bufferlen=+..#BUFFERLEN
        Set maxnodeno=i%MaxNodeNo+($length(buffer)+bufferlen-1\bufferlen),killdest=0
        #; If we have not written any blocks so far check if we should compressIf i%Compress=1,i%MaxNodeNo=0,..NotCompressible(.buffer) Set i%Compress=0
        #; allocate new stream no if neededIf$$$streamTmpGlobalDefault {
            If i%Id="" {
                Set i%Id=$increment(@location)
            } Else {
                Set killdest=1
            }
            if i%%Concurrency {
                set tLockRef = ..%GetLockReference(location,i%Id)
                if tLockRef '= "" {
                    lock +@(tLockRef):$$$LockTimeoutif$test { set tLocked = 1 } else { set tLocked = 0throw##class(%Exception.StatusException).CreateFromStatus($$$ERROR($$$LockFailedToAcquireExclusive,tLockRef)) }
                }
            }
            If killdest Do..BuildValueArray(.subvalue)
        } ElseIf i%%Concurrency {
            #; Make sure we unlock when we are done, lock originally taken out in Write methodset tLockRef = ..%GetLockReference(location,i%Id)
            If tLockRef'=""Set tLocked=1
        }
        Set sid=i%IdIf killdest,$data(@location@(sid)) Kill ^(sid)
        If i%Compress=-1||(i%Compress=2) {
            Set @location@(sid)=maxnodeno,^(sid,0)=i%IOSize
        } Else {
            Set @location@(sid)=maxnodeno_","_i%IOSize_$select(i%Compress=0:"",1:","_i%Compress)
        }
        Set maxnodeno=i%MaxNodeNoSet movedata=0If i%TempGbl'="",$$$streamTmpGlobalDefault {
            Set tmpgbl=i%TempGbl,movedata=1For i=1:1:maxnodeno Set @location@(sid,i)=@tmpgbl@(i)
        }
        
        #; If in WRITE mode have to write out buffer, do not update i%MaxNodeNo here as we keep the temporary streamWhile buffer'="" {
            If$length(buffer)<=bufferlen {
                Set @location@(sid,$increment(maxnodeno))=$$$streamCompressBuf(buffer)
                QUIT
            } Else {
                Set @location@(sid,$increment(maxnodeno))=$$$streamCompressBuf($extract(buffer,1,bufferlen))
                Set buffer=$extract(buffer,bufferlen+1,*)
            }
        }

        #; If we did not move the temp data we have to setup so next write will copy the permanent data to temp storageIf 'movedata {
            #; Clear i%TempGbl so permanent location is not removed on stream closeSet i%TempGbl=""If mode=$$$modeWRITE {
                Set i%Mode=$$$modeWRITEJUSTSAVEDSet i%NodeNo=0,i%Position=1,i%Buffer="",i%AtEnd=0
            } ElseIf mode'=$$$modeREADBUFFER {  ; READBUFFER can leave all settings the same
                #; As data did not move we must update i%MaxNodeNo as this is the max on diskSet i%MaxNodeNo=maxnodeno
                Set i%Mode=$$$modeREADNOTCHANGED
            }
        } Else {
            Set i%Mode=$$$modeWRITEJUSTSAVED
        }

        #; Save the subvalue array for index updatesIf$data(subvalue) Merge ^||%isc.strv(..%Oid())=subvalue

        if tLocked { lock -@(tLockRef)#"I" }
        Do$zu(68,25,batch)
    }
    catch tException {
        if$Get(tLocked) { lock -@(tLockRef)#"I" }
        set tStatus = tException.AsStatus()
    }
    Quit tStatus
}

ClassMethod%DeleteData(streamvalue As%String, concurrency As%Integer) As%Status [ Private ]
{
    try {
        set tStatus=$$$OKset tLocked=0set root=$$$oidSysAd1(streamvalue)
        set shardnum=$$$oidSysAd2(streamvalue)
        if root="" { set root=$$$streamGlobal }
        else {
            set:'$d($system.Context.SQL().ShardNumber($namespace)) $system.Context.SQL().ShardNumber($namespace)=$$getNSShardNumber^%SYS.SHARDSRV()
            if shardnum,shardnum'=$system.Context.SQL().ShardNumber($namespace) { set root="^|"""_$$getExtRefEnvFromShardNum^%SYS.SHARDSRV(shardnum)_"""|"_$e(root,2,*) }
        }
        set node=$$$oidPrimary(streamvalue)
        quit:(root="")||(node="")
        if concurrency {
            set tLockRef = ..%GetLockReference(root,node)
            if tLockRef '= "" {
                lock +@(tLockRef):$$$LockTimeoutif$test { set tLocked = 1 } else { set tStatus=$$$ERROR($$$LockFailedToAcquireExclusive,tLockRef) quit }
            }
        }
        kill @root@(node)
    } catch tException {
        set tStatus = tException.AsStatus()
    }
    lock:tLocked -@(tLockRef)#"I"RETURN tStatus
}

/// Setup the TempGbl location of where we will write the temp stream to
Method SetupTmpGbl() [ Private ]
{
    If i%%Location=""Set i%%Location=$$$streamGlobal
    #; Optimistically write temp stream to permanent destination if location of stream is not the default and this is a new streamIf i%%Location=$$$streamGlobal||(i%Id'="") {
        $$$streamInitTmpGlobal
    } Else {
        Set batch=$zu(68,25,1)
        Set i%Id=$increment(@i%%Location),tLocked=1if i%%Concurrency {
            set tLockRef = ..%GetLockReference(i%%Location,i%Id)
            if tLockRef '= "" {
                lock +@(tLockRef):$$$LockTimeoutSet tLocked=$test
            }
        }
        If 'tLocked {
            $$$streamInitTmpGlobal
        } Else {
            Set i%TempGbl=$name(@i%%Location@(i%Id))
        }
        Do$zu(68,25,batch)
    }
}

Method CopyFromAndSave(source As%Stream.Object) As%Status
{
    Set sc=$$$OK
    #; Optimise for copy from one global to anotherIf$classname(source)=$classname() {
        Set mode=source.Mode
        Set sourceGlvn=$S(mode=$$$modeREADCHANGED||(mode=$$$modeWRITE):source.TempGbl,mode=$$$modeREADNODATA||(mode=$$$modeREADBUFFER):"",1:$na(@(source.%Location)@(source.Id)))
        If mode=$$$modeWRITEJUSTSAVED {
            Set maxnodeno=+$get(@sourceGlvn)
        } Else {
            Set maxnodeno=source.MaxNodeNo
        }
        Set i%Compress=source.Compress
        #; Update the last modified timestampSet i%mLastModified=$ZTimeStamp
        #; Turn on batch mode so reading stream does not kill the cacheSet batch=$zu(68,25,1)
        If i%TempGbl'="" {
            Set journal=$zu(139,2)
            Kill @i%TempGblDo$zu(139,journal)
            If '$$$streamTmpGlobalDefault,i%%Concurrency {
                Set lockref = ..%GetLockReference(i%%Location,i%Id)
                If lockref'=""lock -@(lockref)#"I"
            }
            Set i%TempGbl=""
        }
        #; allocate new stream no if neededSet location=i%%Location,id=i%IdIf id="" {
            Set i%Id=$increment(@i%%Location),id=i%Id
        } Else {
            If$data(@location@(id)) Kill ^(id)
        }
        Set writelocation=$name(@location@(id))
        For i=1:1:maxnodeno Set @writelocation@(i)=@sourceGlvn@(i)
        Set i%MaxNodeNo=maxnodeno,i%IOSize=source.Size,buffer=$select(mode=$$$modeWRITE||(mode=$$$modeREADBUFFER):source.Buffer,1:"")
        #; If we have not written any blocks so far see if it is worth compressingIf i%Compress=1,i%MaxNodeNo=0,..NotCompressible(.buffer) Set i%Compress=0If buffer'="" {
            Set bufferlen=+..#BUFFERLEN
            If$length(buffer)<=bufferlen {
                $$$streamWriteCompressData(buffer,writelocation)
            } Else {
                $$$streamWriteCompressData($extract(buffer,1,bufferlen),writelocation)
                Set buffer=$extract(buffer,bufferlen+1,*)
                While buffer'="" {
                    Set @writelocation@($increment(i%MaxNodeNo))=$$$streamCompressBuf($extract(buffer,1,bufferlen))
                    Set buffer=$extract(buffer,bufferlen+1,*)
                }
            }
        }
        If i%Compress=-1||(i%Compress=2) {
            Set @writelocation=i%MaxNodeNo,^(id,0)=source.Size
        } Else {
            Set @writelocation=i%MaxNodeNo_","_(source.Size)_$select(i%Compress=0:"",1:","_i%Compress)
        }
        Do$zu(68,25,batch)
        Set i%Mode=$$$modeREADNOTCHANGED,i%Buffer=""
        #; Now copy the LineTerminator attributeSet..LineTerminator=source.LineTerminator
        Set i%"%%OID"=..%Oid()
    } Else {
        Set sc=..Rewind() Quit:$$$ISERR(sc) sc
        Set sc=..CopyFrom(source) Quit:$$$ISERR(sc) sc
        Set sc=..%Save()
    }
    Quit sc
}

/// Return the size of the stream pointed to by soidClassMethod%ObjectSizeInternal(soid As%ObjectIdentity) As%Integer [ Private ]
{
    set size=0set location=$$$oidSysAd1(soid)
    set:location="" location=$$$streamGlobalset shardnum=$$$oidSysAd2(soid)
    Set loctmp=$$$oidPrimary(soid)
    RETURN:location=""||(loctmp="") 0set:'$d($system.Context.SQL().ShardNumber($namespace)) $system.Context.SQL().ShardNumber($namespace)=$$getNSShardNumber^%SYS.SHARDSRV()
    if shardnum,shardnum'=$system.Context.SQL().ShardNumber($namespace) { set location="^|"""_$$getExtRefEnvFromShardNum^%SYS.SHARDSRV(shardnum)_"""|"_$e(location,2,*) }
    set rootnode=$get(@location@(loctmp))
    If rootnode["," {
        Set size=$piece(rootnode,",",2)
    } Else {
        #; read size from "0" node, if presentset:$data(@location@(loctmp,0),size)#10=0 size=##class(%Stream.Object)$this.%ObjectSizeInternal(soid)
    }
    RETURN size
}

ClassMethod%LOBPrefetchInternal(soid As%ObjectIdentity, prefetchsize As%Integer, ByRef data As%String) As%Integer [ Private ]
{
    set location=$$$oidSysAd1(soid)
    set:location="" location=$$$streamGlobalset shardnum=$$$oidSysAd2(soid)
    set loctmp=$$$oidPrimary(soid)
    RETURN:location=""||(loctmp="") 0set:'$d($system.Context.SQL().ShardNumber($namespace)) $system.Context.SQL().ShardNumber($namespace)=$$getNSShardNumber^%SYS.SHARDSRV()
    if shardnum,shardnum'=$system.Context.SQL().ShardNumber($namespace) { set location="^|"""_$$getExtRefEnvFromShardNum^%SYS.SHARDSRV(shardnum)_"""|"_$e(location,2,*) }
    set rootnode=$get(@location@(loctmp))
    if rootnode["," {
        Set size=$piece(rootnode,",",2),maxnode=+rootnode
    } Else {
        #; read size from "0" node, if presentRETURN:$data(@location@(loctmp,0),size)#10=0##class(%Stream.Object)$this.%LOBPrefetchInternal(soid,prefetchsize,.data)
        set maxnode=$select(rootnode="":+$order(@location@(loctmp,""),-1),1:+rootnode)
    }
    RETURN:size>prefetchsize 0
    #; Read compression state from header or from class parameterSet compress=$select(rootnode[",":$piece(rootnode,",",3),1:..#COMPRESS)
    for i=1:1:maxnode {
        set data=data_$$$streamDecompressBufFlag(@location@(loctmp,i),compress)
    }
    RETURN$select(size=0:3,..IsCharacter():1,1:2)
}

Storage Default
{
<Data name="GlobalBinaryDefaultData">
<Value name="1">
<Value>%%CLASSNAME</Value>
</Value>
</Data>
<DataLocation>^%Stream.GlobalBinaryD</DataLocation>
<DefaultData>GlobalBinaryDefaultData</DefaultData>
<IdLocation>^%Stream.GlobalBinaryD</IdLocation>
<IndexLocation>^%Stream.GlobalBinaryI</IndexLocation>
<StreamLocation>^%Stream.GlobalBinaryS</StreamLocation>
<Type>%Storage.Persistent</Type>
}

}