Repository: gojue/ecapture
File: pkg/event_processor/iworker.go (Go)

Code before:
		return
	}
	ew.payload.Write(e.Payload())
}

// 解析类型，输出
func (ew *eventWorker) parserEvents() []byte {
	ew.status = ProcessStateProcessing
	parser := NewParser(ew.payload.Bytes())
	ew.parser = parser
	n, e := ew.parser.Write(ew.payload.Bytes())
	if e != nil {
		_ = ew.writeToChan(fmt.Sprintf("ew.parser write payload %d bytes, error:%v", n, e))
	}
	ew.status = ProcessStateDone
	return ew.parser.Display()
}

func (ew *eventWorker) Run() {
	tsize := int(ew.processor.truncateSize)
	for {
		select {
		case <-ew.ticker.C:
			truncateFlag := (tsize > 0 && ew.payload.Len() >= tsize)
			// 输出包
			if truncateFlag || ew.tickerCount > MaxTickerCount {
				if truncateFlag {
					ew.payload.Truncate(tsize)
					_ = ew.writeToChan(fmt.Sprintf("Events truncated, size: %d bytes\n", tsize))
				}
				//ew.processor.GetLogger().Printf("eventWorker TickerCount > %d, event closed.", MaxTickerCount)
				ew.processor.delWorkerByUUID(ew)

				/*
					When returned from delWorkerByUUID(), there are two possibilities:
					1) no routine can touch it.
					2) one routine can still touch ew because getWorkerByUUID()
					*happen before* delWorkerByUUID()

					When no routine can touch it (i.e.,ew.IfUsed == false),
					we just drain the ew.incoming and return.

					When one routine can touch it (i.e.,ew.IfUsed == true), we ensure
					that we only return after the routine can not touch it
					(i.e.,ew.IfUsed == false). At this point, we can ensure that no
					other routine will touch it and send events through the ew.incoming.
					So, we return.

					Because eworker has been deleted from workqueue after delWorkerByUUID()
					(ordered by a workqueue lock), at this point, we can ensure that
					no ew will not be touched even **in the future**. So the return is

Review comment (refactor):
I think it is better to move this part into `func (ew *eventWorker) writeEvent(e event.IEventStruct)`.

Code after:
		return
	}

	tsize := int(ew.processor.truncateSize)
	//terminal write when reach the truncate size
	if tsize > 0 && ew.payload.Len() >= tsize {
		ew.payload.Truncate(tsize)
		return
	}

	ew.payload.Write(e.Payload())
}

// 解析类型，输出
func (ew *eventWorker) parserEvents() []byte {
	ew.status = ProcessStateProcessing
	parser := NewParser(ew.payload.Bytes())
	ew.parser = parser
	n, e := ew.parser.Write(ew.payload.Bytes())
	if e != nil {
		_ = ew.writeToChan(fmt.Sprintf("ew.parser write payload %d bytes, error:%v", n, e))
	}
	ew.status = ProcessStateDone
	return ew.parser.Display()
}

func (ew *eventWorker) Run() {
	for {
		select {
		case <-ew.ticker.C:
			// 输出包
			if ew.tickerCount > MaxTickerCount {
				//ew.processor.GetLogger().Printf("eventWorker TickerCount > %d, event closed.", MaxTickerCount)
				ew.processor.delWorkerByUUID(ew)

				/*
					When returned from delWorkerByUUID(), there are two possibilities:
					1) no routine can touch it.
					2) one routine can still touch ew because getWorkerByUUID()
					*happen before* delWorkerByUUID()

					When no routine can touch it (i.e.,ew.IfUsed == false),
					we just drain the ew.incoming and return.

					When one routine can touch it (i.e.,ew.IfUsed == true), we ensure
					that we only return after the routine can not touch it
					(i.e.,ew.IfUsed == false). At this point, we can ensure that no
					other routine will touch it and send events through the ew.incoming.
					So, we return.

					Because eworker has been deleted from workqueue after delWorkerByUUID()