Accumulated changes for message queue (#6600)

* rename

* set agent address

* refactor

* add agent sub

* pub messages

* grpc new client

* can publish records via agent

* send init message with session id

* fmt

* check cancelled request while waiting

* use sessionId

* handle possible nil stream

* subscriber process messages

* separate debug port

* use atomic int64

* less logs

* minor

* skip io.EOF

* rename

* remove unused

* use saved offsets

* do not reuse session, since always session id is new after restart

remove last active ts from SessionEntry

* simplify printing

* purge unused

* just proxy the subscription, skipping the session step

* adjust offset types

* subscribe offset type and possible value

* start after the known tsns

* avoid wrongly set startPosition

* move

* remove

* refactor

* typo

* fix

* fix changed path

1 files changed, 4 insertions, 4 deletions

diff --git a/weed/util/log_buffer/log_buffer.go b/weed/util/log_buffer/log_buffer.go
index 30498f92d..fb1f8dc2f 100644
--- a/weed/util/log_buffer/log_buffer.go
+++ b/weed/util/log_buffer/log_buffer.go
@@ -46,7 +46,7 @@ type LogBuffer struct {
 	isStopping        *atomic.Bool
 	isAllFlushed      bool
 	flushChan         chan *dataToFlush
-	LastTsNs          int64
+	LastTsNs          atomic.Int64
 	sync.RWMutex
 }
 
@@ -95,12 +95,12 @@ func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processin
 	} else {
 		ts = time.Unix(0, processingTsNs)
 	}
-	if logBuffer.LastTsNs >= processingTsNs {
+	if logBuffer.LastTsNs.Load() >= processingTsNs {
 		// this is unlikely to happen, but just in case
-		processingTsNs = logBuffer.LastTsNs + 1
+		processingTsNs = logBuffer.LastTsNs.Add(1)
 		ts = time.Unix(0, processingTsNs)
 	}
-	logBuffer.LastTsNs = processingTsNs
+	logBuffer.LastTsNs.Store(processingTsNs)
 	logEntry := &filer_pb.LogEntry{
 		TsNs:             processingTsNs,
 		PartitionKeyHash: util.HashToInt32(partitionKey),