ProtoTree x->right ==x Caused infinite loop about norm HOT 41 OPEN

import os
dirPath = r'E:\PythonPrj/testFiles'

if not os.path.isdir(dirPath):
    os.mkdir(dirPath)
for i in range(1, 10000):
    with open(os.path.join(dirPath, f"文件{i:04d}.txt"), "w") as f:
        f.write(f"{i} "*5_000)

async def proc_sender_event(event:pynorm.event.Event):
    #print(f"{str(event)}")
    obj = event.object
    evtType:pynorm.EventType = event.type
    session:pynorm.session.Session = event.session
    linkID:int = session.conf['linkID']
    userName:str = session.conf['userName']
    
    if obj:   #and obj.handle not in (None,pynorm.NORM_OBJECT_INVALID):
        objType:ObjectType = obj.type
        info:str|None =obj.info.decode() if obj.info else None
        print(event, linkID, obj.type, info)
    else:
        print (event)
    
    if evtType == EventType.TX_OBJECT_SENT:
        if objType == ObjectType.FILE and obj.info:
            session.sendFiles.add(obj.info.decode())
            
    elif evtType == EventType.TX_QUEUE_EMPTY:
        '''
        '''
        for key, obj in session.id2obj.items():
            obj.cancel()
        session.id2obj.clear()
        #用于关闭obj的文件句柄
        del obj
        del event 
        
        #listProcOpenFiles() #用于测试是否有未关闭的文件句柄 
        for file in session.sendFiles:
            try:
                moveFile2SendedDir(file)
            except Exception as e:
                logging.warning( traceback.format_exc() )
        session.sendFiles.clear()
        files = await get_some_waitingfiles_of_link(settings.sendFileDir,linkID)
        pushFiles2session(session, waitingFiles=files)
        
    elif evtType == EventType.TX_OBJECT_PURGED:
        '''
            
        '''
        if obj.handle in session.id2obj:
            session.id2obj.pop(obj.handle)
        info = obj.info.decode()
        if info in session.sendFiles:
            session.sendFiles.remove( info )
        obj.cancel()
        del obj
        del event
        try:
            moveFile2SendedDir( info )
        except Exception as e:
            logging.warning( traceback.format_exc() )
            
        
    elif evtType == EventType.TX_FLUSH_COMPLETED:
        '''
            the NORM sender observes when it no longer has data ready for transmission has completed
        '''
        for file in session.sendFiles:
            moveFile2SendedDir(file)
        session.sendFiles.clear()        
        files = await get_some_waitingfiles_of_link(settings.sendFileDir,linkID)
        pushFiles2session(session, waitingFiles=files)
        
    elif event.type == EventType.TX_OBJECT_PURGED:
        
        # 参见 NormSetTxCacheBounds 说明 
        handle:int = obj.handle
        if objType and handle in event.session.id2obj:
            event.session.id2obj.pop(handle) 
            
        if objType == ObjectType.FILE and obj.info:
            '''
                move sended files to the sended dir
            '''
            info:str = obj.info.decode()
            obj.cancel()
            del obj
            del event
            moveFile2SendedDir(info)
            if info in session.sendFiles:
                session.sendFiles.remove(info)
            filePath = await get_one_waitingfile_of_link(settings.sendFileDir, session.conf['linkID'])
            pushFiles2session(session, waitingFiles=[filePath])

async def proc_recver_file_event(event:pynorm.event.Event):
    '''
         file info is {channelName}/{fileName} encoding: uft-8 
    '''
    #mqttClient:MQTTClient = mqtt_client()
    
    evtType:EventType = event.type
    session:pynorm.session.Session = event.session
    obj:pynorm.object.Object = event.object
    #linkID = session.conf['linkID']
    info:str = obj.info.decode() if obj and obj.info else None
    print (f" {evtType}  {info=}")

    if evtType == EventType.RX_OBJECT_NEW:
        now:str = datetime.datetime.now().isoformat()
        
        channelName, fileName = info.split("/")
        obj.begin = now
        obj.fileName = fileName
        obj.channelName = channelName
        session.id2obj[ info ] =  obj


    elif evtType == EventType.RX_OBJECT_COMPLETED:
        '''
        This event is posted when a receive object is completely received, including available NORM_INFO content. 
        Unless the application specifically retains the "object" handle, the indicated NormObjectHandle becomes invalid and must no longer be referenced.
        '''
        if info in session.id2obj:
            session.id2obj.pop( info )
        #obj.cancel()
        
        newFilePath:str = os.path.join(settings.recvFileDir, obj.channelName, obj.fileName )
        oldPath:str = event.object.filename.decode( locale.getpreferredencoding() )
        print (f"{oldPath=} {newFilePath}")
        try:
            #remove conflict file 
            if os.path.isfile(newFilePath):
                os.remove(newFilePath)
            newDirPath:str = os.path.join(settings.recvFileDir, obj.channelName )
            #create dest dir.
            if not os.path.isdir(newDirPath):
                os.makedirs( newDirPath )           
            #remove temp file to new dir and give the final real name. 
            os.rename(src=oldPath, dst=newFilePath)
        except Exception as e:
            logging.warning ( traceback.format_exc() )  
            
        logging.info(f"{obj.channelName, obj.fileName} completed")
        
    elif evtType == EventType.RX_OBJECT_ABORTED:
        oldPath:str = event.object.filename.decode( locale.getpreferredencoding() )
        obj.cancel()
        os.remove(oldPath) #移除 临时文件 

        logging.warning( f'{obj.channelName, obj.fileName} aborted' )    

      elif evtType == EventType.TX_FLUSH_COMPLETED:
        '''
            the NORM sender observes when it no longer has data ready for transmission has completed
        '''
        for key, obj in session.id2obj.items():
            obj.cancel()
        session.id2obj.clear()
        del obj
        
        for file in session.sendFiles:
            moveFile2SendedDir(file)
        session.sendFiles.clear()   
 
        files = await get_some_waitingfiles_of_link(settings.sendFileDir,linkID)
        pushFiles2session(session, waitingFiles=files)

del obj
del event

    elif event.type == EventType.TX_OBJECT_PURGED:
        # 参见 NormSetTxCacheBounds 说明 
        handle:int = obj.handle
        if objType and handle in event.session.id2obj:
            event.session.id2obj.pop(handle) 
            
        if objType == ObjectType.FILE and obj.info:
            info:str = obj.info.decode()
            #obj.cancel()
            #del obj
            #del event
            #moveFile2SendedDir(info)
            if info in session.sendFiles:
                session.sendFiles.remove(info)
            filePath = await get_one_waitingfile_of_link(settings.sendFileDir, linkID, not_in=session.sendFiles)
            pushFiles2session(session, waitingFiles=[filePath])
            loop = asyncio.get_running_loop()
            loop.call_later(1, moveFile2SendedDir, info)  

elif event.type == EventType.TX_OBJECT_PURGED:
    if info in session.info2obj:
        session.info2obj.pop(info)

    # 1, to avoid same file enqueued twice.
    #if info in session.sendFiles:
        #session.sendFiles.remove( info )        
        
    if objType == ObjectType.FILE and obj.info:
        filePath = await get_one_waitingfile_of_link(settings.sendFileDir, linkID, not_in=session.sendFiles)
        pushFiles2session(session, waitingFiles=[filePath])
        
        #2, waiting 30seconds for NORM to close file handle. 
        loop = asyncio.get_running_loop()
        loop.call_later(30, moveFile2SendedDir, info) 

if info in session.info2obj:
    session.info2obj.pop(info)
      
if objType == ObjectType.FILE and obj.info:
    moveFile2SendedDir(info)
    
    if info in session.sendFiles:
        session.sendFiles.remove( info )
    
    filePath = await get_one_waitingfile_of_link(settings.sendFileDir, linkID, not_in=session.sendFiles)
    pushFiles2session(session, waitingFiles=[filePath])

    for i in range(1, 100_000):
        with open(os.path.join(dirPath, f"文件{i:05d}.txt"), "w") as f:
            f.write(f"{i} "*5_000)    

import os
if os.name=='nt':
    #os.add_dll_directory(os.path.dirname(__file__))
    os.add_dll_directory(r'E:\PythonPrj\NORM\norm159\makefiles\win32\x64\Release')
import asyncio
import logging
import socket
import traceback
import re

import win32event
import netifaces

def get_interface_by_ip(ip:str):
    '''
       根据IP地址获取对应的接口，
       耗时Win7 0.07秒  Deepin/SW 0.001秒 
    '''
    for iface in netifaces.interfaces():
        allAddrs = netifaces.ifaddresses(iface)        
        if addrs:= allAddrs.get(socket.AF_INET):
            for addr in addrs:
                if addr['addr'] == ip:
                    return iface
        elif addrs:= allAddrs.get(socket.AF_INET6):
            for addr in addrs:
                if addr['addr'] == ip:
                    return iface
import pynorm

def pushFiles2session(session:pynorm.session.Session, waitingFiles:list[str], watermark=False) -> list[str]:
    '''
        try enqueue some files to `session`, return those succeed enqueued ones,
        向Session添加若干待发送文件,返回成功加入到队列的文件 
    '''
    added_files =[]
    for index, filePath in enumerate(waitingFiles):
        channelName, fileName = filePath.split("/")
        fullFilePath:str = os.path.join(settings.sendFileDir, channelName, fileName )
        
        #the session sending queue already has same filePath.  跳过队列中已有的同名文件,
        if filePath in session.info2obj:
            continue 
            #obj = session.info2obj.pop(filePath)
            #obj.cancel()
        obj:pynorm.Object = session.fileEnqueue( fullFilePath , info=filePath.encode() )
        handle:int|None = obj.handle
        if handle in (None, pynorm.NORM_OBJECT_INVALID):
            logging.info(f'pushed files:{added_files}')
            return added_files
        else:
            added_files.append( filePath )
            session.info2obj[filePath] = obj 
        if watermark:
            session.setWatermark(obj)
       
    logging.info(f'pushed files:{added_files}')
    return added_files

def moveFile2SendedDir(info:str) -> bool:
    currentFilePath = os.path.join(settings.sendFileDir, info)
    if not os.path.isfile(currentFilePath):
        return False
        #raise FileNotFoundError(currentFilePath)
    channelName, fileName = info.split('/')
    #1, create dir if not exits, 没有则创建目录
    dirPath = os.path.join(settings.sendedFileDir, channelName )
    if not os.path.exists(dirPath):
        os.makedirs(dirPath)
    newFilePath = os.path.join(dirPath, fileName)
    
    # 移除重名的已下载文件
    if os.path.isfile(newFilePath):
        os.remove(newFilePath)
    
    os.rename( currentFilePath, newFilePath )  #移动到已发送目录    
    logging.info( f"{info} moved！")
    return True

def moveFiles2SendedDir(infos:list[str])-> bool:
    for info in infos:
        moveFile2SendedDir(info)
        

def _get_some_waitingfiles_of_link(basicPath:str, channnelName:str, limit:int=20, not_in:dict|None=None):
    '''
        获取一个频道的待发送文件列表,
        limit: 限制最大数量
    '''
    path:str = os.path.join(basicPath, channnelName)
    #没有创建频道目录，
    if not os.path.isdir(path):
        #os.makedirs(path)
        return []
    
    waiting_files=[]
    dir_or_files = os.listdir(path)
    fileNum =0 
    for fileName in dir_or_files:
        file_path = os.path.join(path,fileName)
        # 临时文件、已完成文件不发送 
        if os.path.isfile(file_path) and not re.findall("\.tmp$|\.end$", fileName): 
            filePath =  f"{channnelName}/{fileName}"
            if not_in is None or filePath not in not_in:   # 未设置或者 文件名称不在待处理的名单中              
                waiting_files.append(filePath)
                fileNum +=1
                if fileNum >=limit:
                    break   
    return waiting_files

from aiofiles.os import wrap

get_some_waitingfiles_of_link = wrap(_get_some_waitingfiles_of_link)



import ipaddress
import typing
from pydantic import BaseSettings
class Settings(BaseSettings):
    sendFileDir:str=""
    sendedFileDir:str =""
    
settings = Settings()

import pynorm    
from pynorm.constants import EventType,NackingMode

SENDER_BUFFER_SPACE = 100*1024*1024

#from util.netifaces_ext import get_interface_by_ip
def create_sender_session(instance:pynorm.Instance, 
                          destAddr:str,
                          destPort:int, 
                          localAddr:str|None=None, 
                          localPort:int=0,                            
                          iface:str|None=None, # the interface of <localAddr>
                          srcAddr:str|None=None, #usef for setSSM
                          sessionIndex:typing.Hashable=None,
                          ccEnable:bool = True,
                          rateMin:float|None=-1,
                          rateMax:float|None=-1,
                          txRate:float|None =None,
                          bufferSpace:int = SENDER_BUFFER_SPACE,
                          segmentSize:int = 1400,
                          blockSize:int = 128,
                          numParity:int = 0,
                          loopbackEnable:bool = False
                        
                          ) -> pynorm.session.Session:

    session = instance.createSession(destAddr, destPort,  localId=ipaddress.IPv4Address(localAddr)._ip, index=sessionIndex  )
    if session._session == pynorm.NORM_SESSION_INVALID:
        raise Exception( f"createSession: NORM_SESSION_INVALID {destAddr}")
    if localAddr or localPort:
        session.setTxPort(localPort,txBindAddr=localAddr) #
    if iface:
        session.setMulticastInterface(iface)
    if srcAddr:
        session.setSSM(srcAddr=srcAddr)    
    session.setTxOnly(txOnly=True) 
    if txRate:
        session.setTxRate(txRate*1000)
    session.setCongestionControl(ccEnable=ccEnable) # 
    if loopbackEnable:
        session.setLoopback(True)
    if rateMin>=0 or rateMax>=0:
        session.setTxRateBounds(rateMin=rateMin*1000,rateMax=rateMax*1000)

    #sessionID = randint(0, 1000)
    success:bool = session.startSender(sessionIndex, bufferSpace, segmentSize=segmentSize, blockSize=blockSize, numParity=numParity ) 
    print (f"startSender:{success}")
    #session.setGroupSize(4)
    return session 
                
                
instance:pynorm.Instance = pynorm.Instance()
async def start_one_link(sendFileDir:str, link:dict, pushFileNum:int=10):
    global instance
    localAddr =link['localAddr'] 
    iface = link.get('iface')
    if os.name =='posix' and iface is None:
        loop = asyncio.get_running_loop()
        iface:str|None  = await loop.run_in_executor(None, get_interface_by_ip, localAddr)
        if not iface:
            raise Exception(f"{localAddr} is not valid local Addr")   
        
    session = create_sender_session(instance=instance,
                          destAddr=link['privateMulticastAddr'], destPort=link['destPort'],
                          localAddr = localAddr ,
                          #localPort:int=0,                            
                          iface= iface, # the interface of <localAddr>
                          #srcAddr:str|None=None, #usef for setSSM
                          sessionIndex = link['linkID'],
                          ccEnable = link['ccEnable'],
                          rateMin =link['rateMin'],
                          rateMax =link['rateMax'],
                          txRate =link['txRate'], 
                          bufferSpace = 100*1024*1024,
                          segmentSize = link['segmentSize'],
                          blockSize = link['blockSize'],
                          numParity = link['numParity'],
                          loopbackEnable=False,
                          )
    session.conf= link
    
    if ttl:=link.get('TTL'):
        session.setTTL(ttl)
    if tos:=link.get('TOS'):
        session.setTOS(tos)
    if autoParity:= link.get("autoParity"):
        session.setAutoParity(autoParity) 
    if waitingFiles:= link.get('waitingFiles'):
        pushFiles2session(session, waitingFiles[:pushFileNum])
        
    session.setTxCacheBounds(50_000_000, 200, 200) #NormSetTxCacheBounds
    files = await  get_some_waitingfiles_of_link(settings.sendFileDir,channnelName=link['channels'][0], )
    pushFiles2session(session, waitingFiles=files)
    

from pynorm.constants import  ObjectType

import copy           
async def proc_sender_event(event:pynorm.Event):
    '''
    
    '''
    #print(f"{str(event)}")
    obj = event.object
    evtType:pynorm.EventType = event.type
    session:pynorm.Session = event.session
    linkID:int = session.conf['linkID']
    
    if obj:
        objType:ObjectType = obj.type
        info:str|None =obj.info.decode() if obj.info else None
        print(event, linkID, obj.type, info)
    else:
        pass
        #print (event)
    
    if evtType == EventType.TX_OBJECT_SENT:
        '''
            需要缓存已经发送的对象，当收到TX_FLUSH_COMPLETED或TX_OBJECT_PURGED或者TX_QUEUE_EMPTY时  
        '''
        pass
        
    elif event.type == EventType.TX_OBJECT_PURGED:

            
        ## 不删除避免同一文件重复添加, to avoid same file enqueued twice.
        if objType == ObjectType.FILE:
            loop=asyncio.get_running_loop()
            #_future:asyncio.Future  = loop.run_in_executor(None, moveFile2SendedDir, info)  
            # 问题： 不能直接启动，在移动完成前从info2obj中移除， 这会导致重复添加文件  
            await loop.run_in_executor(None, moveFile2SendedDir, info) 
            if info in session.info2obj:
                session.info2obj.pop(info)        
                
                
    elif evtType == EventType.TX_QUEUE_EMPTY:
        if len(session.info2obj)>200:
            return 
        files = await  get_some_waitingfiles_of_link(settings.sendFileDir, channnelName=session.conf['channels'][0], limit=10, not_in=session.info2obj )
        if files:
            pushFiles2session(session, waitingFiles=files)
        
    elif evtType == EventType.TX_FLUSH_COMPLETED:
        '''
            the NORM sender observes when it no longer has data ready for transmission has completed
        '''
        try:
            
            files =await get_some_waitingfiles_of_link(settings.sendFileDir,channnelName=session.conf['channels'][0],  not_in=session.info2obj)
            if files:
                pushFiles2session(session, waitingFiles=files)
            else:
                # 当没有新文件需要发送时， 将队列中的文件移出到已发送目录 
                infos =[]
                for info, obj in session.info2obj.items():
                    obj.cancel()
                    infos.append(info)
                session.info2obj.clear()
                loop=asyncio.get_running_loop()
                _future:asyncio.Future  = loop.run_in_executor(None, moveFiles2SendedDir, infos)
        except Exception as e:
            print (traceback.format_exc())
        
import typing
from tornado.ioloop import PeriodicCallback

import pynorm
import select
import time
from pynorm import DebugLevel as NormDebugLevel
import logging
async def watch_norm_events(proc_event: typing.Callable[[pynorm.event.Event],None ], timeout:int=1):
    global instance
    instance.setDebugLevel(level=NormDebugLevel.INFO) #2 Warning 3 INFO 4 DEBUG


    handle = instance.getDescriptor()
    while True:
        try:
            if os.name =='nt':
                '''
                    DWORD WaitForSingleObject(
                     [in] HANDLE hHandle,
                     [in] DWORD  dwMilliseconds
                   );
                   [in] dwMilliseconds

                   The time-out interval, in milliseconds. If a nonzero value is specified, 
                   the function waits until the object is signaled or the interval elapses. 
                   If dwMilliseconds is zero, the function does not enter a wait state if the object is not signaled; 
                   it always returns immediately.
                   If dwMilliseconds is INFINITE, the function will return only when the object is signaled.

                '''
                value:int = await asyncio.to_thread(win32event.WaitForSingleObject, handle,1_000 ) # win32event.INFINITE
                if value == win32event.WAIT_TIMEOUT:
                    await asyncio.sleep(0)
                    continue
                elif value == win32event.WAIT_FAILED:
                    print ( f"error:{win32api.GetLastError()}" )
                elif value == win32event.WAIT_ABANDONED:
                    pass 
                elif value == win32event.WAIT_OBJECT_0:       
                    while event:= instance.getNextEvent( ):
                        beg = time.time()
                        await proc_event(event)
                        print ( f" {event.type.name} {time.time()-beg:.4f}")
                    #print(event) 
            else:
                readable, writable, exceptional = await asyncio.to_thread(select.select, [handle],[],[handle]) 
                if readable:
                    event: pynorm.event.Event = instance.getNextEvent( )
                    await proc_event(event)              
                    
        except Exception as e:
            logging.error ( traceback.format_exc() )
    print("watch_norm_events finished" )
    return 0
        
        
async def periodic_func():
    for session in instance._sessions.values():
        print (f"{session.multiAddr},  {session.getSenderReport().get_dict()}")
   
import tornado 
if __name__ =='__main__':
    logging.getLogger().setLevel(logging.INFO)
    link =  { 'linkID': 1, 'userID': 1, 'linkName': 'a', 'destPort': 6003, 'localAddr': '10.65.39.191',
              'TTL': None, 'TOS': None, 'ccEnable': False,
              'rateMin': -1, 'rateMax': 8000000, 'txRate': 8000000,
              'segmentSize': 1400, 'blockSize': 128, 'numParity': 4, 'autoParity': 1,
              'waitingFiles': [], 'channels': ['a中文频道1'],
              'sendFileDir': '../sendFiles',
              'privateMulticastAddr': '224.1.2.4'
              }

    link2 =  { 'linkID': 1, 'userID': 1, 'linkName': 'a', 'destPort': 6003, 'localAddr': '10.65.39.191',
              'TTL': None, 'TOS': None, 'ccEnable': False,
              'rateMin': -1, 'rateMax': 8000000, 'txRate': 8000000,
              'segmentSize': 1400, 'blockSize': 128, 'numParity': 4, 'autoParity': 1,
              'waitingFiles': [], 'channels': ['频道2'],
              'sendFileDir': '../sendFiles',
              'privateMulticastAddr': '224.1.2.3'
              }
    
    
    
    settings.sendFileDir= '../sendFiles'  # read files from subdir, each subdir is a channel 
    settings.sendedFileDir = '../sendedFiles'
    #loop = asyncio.get_event_loop_policy().new_event_loop() 
    t_loop = tornado.ioloop.IOLoop.current()
    loop = t_loop.asyncio_loop
    
    periodic_check:PeriodicCallback = PeriodicCallback( periodic_func , callback_time=10_000) 
    periodic_check.start()
    
    loop.create_task( watch_norm_events( proc_sender_event) )
    asyncio.run(  start_one_link( settings.sendFileDir, link) )
    asyncio.run(  start_one_link( settings.sendFileDir, link2) )
    loop.run_forever()