// MessageWriteLoop processes all write requests
MessageWriteLoop processes all write requests
// MessageWriteLoop processes all write requests
func (mh *MessageHandle) MessageWriteLoop(hi hubio.CloudHubIO, info *model.HubInfo, stop chan ExitCode) {
	messages, err := mh.MessageQueue.Consume(info)
	if err != nil {
		klog.Errorf("failed to consume event for node %s, reason: %s", info.NodeID, err.Error())
		stop <- messageQueueDisconnect
		return
	}
	for {
		msg, err := messages.Get()
		if err != nil {
			klog.Errorf("failed to consume event for node %s, reason: %s", info.NodeID, err.Error())
			if err.Error() == MsgFormatError {
				messages.Ack()
				continue
			}
			stop <- messageQueueDisconnect
			return
		}
		if model.IsNodeStopped(msg) {
			klog.Infof("node %s is stopped, will disconnect", info.NodeID)
			messages.Ack()
			stop <- nodeStop
			return
		}
		if !model.IsToEdge(msg) {
			klog.Infof("skip only to cloud event for node %s, %s, content %s", info.NodeID, dumpMessageMetadata(msg), msg.Content)
			messages.Ack()
			continue
		}
		klog.Infof("event to send for node %s, %s, content %s", info.NodeID, dumpMessageMetadata(msg), msg.Content)
		trimMessage(msg)
		err = hi.SetWriteDeadline(time.Now().Add(time.Duration(mh.WriteTimeout) * time.Second))
		if err != nil {
			klog.Errorf("SetWriteDeadline error, %s", err.Error())
			stop <- hubioWriteFail
			return
		}
		err = mh.hubIoWrite(hi, info.NodeID, msg)
		if err != nil {
			klog.Errorf("write error, connection for node %s will be closed, affected event %s, reason %s",
				info.NodeID, dumpMessageMetadata(msg), err.Error())
			stop <- hubioWriteFail
			return
		}
		messages.Ack()
	}
}
func ( mh * MessageHandle ) MessageWriteLoop ( hi hubio . CloudHubIO , info * model . HubInfo , stop chan ExitCode ) { messages , err := mh . MessageQueue . Consume ( info ) if err != nil { klog . Errorf ( " " , info . NodeID , err . Error ( ) ) stop <- messageQueueDisconnect return } for { msg , err := messages . Get ( ) if err != nil { klog . Errorf ( " " , info . NodeID , err . Error ( ) ) if err . Error ( ) == MsgFormatError { messages . Ack ( ) continue } stop <- messageQueueDisconnect return } if model . IsNodeStopped ( msg ) { klog . Infof ( " " , info . NodeID ) messages . Ack ( ) stop <- nodeStop return } if ! model . IsToEdge ( msg ) { klog . Infof ( " " , info . NodeID , dumpMessageMetadata ( msg ) , msg . Content ) messages . Ack ( ) continue } klog . Infof ( " " , info . NodeID , dumpMessageMetadata ( msg ) , msg . Content ) trimMessage ( msg ) err = hi . SetWriteDeadline ( time . Now ( ) . Add ( time . Duration ( mh . WriteTimeout ) * time . Second ) ) if err != nil { klog . Errorf ( " " , err . Error ( ) ) stop <- hubioWriteFail return } err = mh . hubIoWrite ( hi , info . NodeID , msg ) if err != nil { klog . Errorf ( " " , info . NodeID , dumpMessageMetadata ( msg ) , err . Error ( ) ) stop <- hubioWriteFail return } messages . Ack ( ) } }
MessageWriteLoop processes all write requests
MessageWriteLoop processes all write requests
// error format message should not impact other message