func (t *simpleQueue) backgroundtask() {
	var retryqueue list.List // list of queued messages; the list is not multithreading safe, and must only be accessed from backgroundtask()

	goglog.Logger.Debug("backgroundtask started for ", t.GetType())
	dur := time.Duration(t.RetryInterval) * time.Second
	ticker := time.NewTicker(dur)
	defer ticker.Stop()
	for {
		select {
		case event := <-t.queue:
			if (retryqueue.Len() < t.MaxQueueSize) || t.MaxQueueSize == -1 {
				retryqueue.PushBack(event)
			}
		case <-t.ctx.Done():
			goglog.Logger.Debugf("queue %s closing", t.GetType())
			return
		case <-ticker.C:
			// We have reached a RetryInterval. If there are any events in the queue, lets send one back.
			// If we are still in pause mode we will send one, if we are in normal mode we will send all back.
			if e := retryqueue.Front(); e != nil {
				goglog.Logger.Debugf("%s simplequeue trying to deliver, has %v events in queue", t.GetType(), retryqueue.Len())
				if atomic.LoadUint32(&t.isInPause) == StatusPaused {
					msg := e.Value
					retryqueue.Remove(e)
					switch v := msg.(type) {
					case logevent.LogEvent:
						go func() {
							ctx, cancel := context.WithTimeout(context.Background(), dur)
							err := t.output.OutputEvent(ctx, v)
							if err != nil {
								goglog.Logger.Errorf("queue: %ssendone %s", t.GetType(), err.Error())
							}
							cancel()
						}()
					case []byte:
						go func() {
							select {
							case <-t.ctx.Done():
								goglog.Logger.Debugf("queue %s closing", t.GetType())
								return
							case t.codecCh <- v:
							}
						}()
					default:
						goglog.Logger.Errorf("Invalid type %T", v)
					}
				} else {
					// we are not in pause mode and will queue all the events in the queue for sending.
					// First we need to empty the queue and get all events to send.
					myList := []interface{}{}
					for {
						e := retryqueue.Front()
						if e == nil {
							break
						}
						event := e.Value
						myList = append(myList, event)
						retryqueue.Remove(e)
					}
					// Now we have to send them all out
					go func() {
						for x := range myList {
							ctx, cancel := context.WithTimeout(context.Background(), dur)
							defer cancel()
							switch v := (myList[x]).(type) {
							case []byte:
								select {
								case <-t.ctx.Done():
									return
								case t.codecCh <- v:
								}
							case logevent.LogEvent:
								err := t.Output(ctx, v)
								if err != nil {
									goglog.Logger.Errorf("queue: %s sendall %s", t.GetType(), err.Error())
								}
							default:
								goglog.Logger.Errorf("Invalid type %T", v)
							}
						}
					}()
				}
			}
		}
	}
}
// backgroundtask is running in the background and adds new events to the queue, tries to send them out on the RetryInterval.
backgroundtask is running in the background and adds new events to the queue, tries to send them out on the RetryInterval.
// backgroundtask is running in the background and adds new events to the queue tries to send them out on the RetryInterval .
func (t *simpleQueue) backgroundtask() {
	var retryqueue list.List 
	goglog.Logger.Debug("backgroundtask started for ", t.GetType())
	dur := time.Duration(t.RetryInterval) * time.Second
	ticker := time.NewTicker(dur)
	defer ticker.Stop()
	for {
		select {
		case event := <-t.queue:
			if (retryqueue.Len() < t.MaxQueueSize) || t.MaxQueueSize == -1 {
				retryqueue.PushBack(event)
			}
		case <-t.ctx.Done():
			goglog.Logger.Debugf("queue %s closing", t.GetType())
			return
		case <-ticker.C:
			if e := retryqueue.Front(); e != nil {
				goglog.Logger.Debugf("%s simplequeue trying to deliver, has %v events in queue", t.GetType(), retryqueue.Len())
				if atomic.LoadUint32(&t.isInPause) == StatusPaused {
					msg := e.Value
					retryqueue.Remove(e)
					switch v := msg.(type) {
					case logevent.LogEvent:
						go func() {
							ctx, cancel := context.WithTimeout(context.Background(), dur)
							err := t.output.OutputEvent(ctx, v)
							if err != nil {
								goglog.Logger.Errorf("queue: %ssendone %s", t.GetType(), err.Error())
							}
							cancel()
						}()
					case []byte:
						go func() {
							select {
							case <-t.ctx.Done():
								goglog.Logger.Debugf("queue %s closing", t.GetType())
								return
							case t.codecCh <- v:
							}
						}()
					default:
						goglog.Logger.Errorf("Invalid type %T", v)
					}
				} else {
					myList := []interface{}{}
					for {
						e := retryqueue.Front()
						if e == nil {
							break
						}
						event := e.Value
						myList = append(myList, event)
						retryqueue.Remove(e)
					}
					go func() {
						for x := range myList {
							ctx, cancel := context.WithTimeout(context.Background(), dur)
							defer cancel()
							switch v := (myList[x]).(type) {
							case []byte:
								select {
								case <-t.ctx.Done():
									return
								case t.codecCh <- v:
								}
							case logevent.LogEvent:
								err := t.Output(ctx, v)
								if err != nil {
									goglog.Logger.Errorf("queue: %s sendall %s", t.GetType(), err.Error())
								}
							default:
								goglog.Logger.Errorf("Invalid type %T", v)
							}
						}
					}()
				}
			}
		}
	}
}
func ( t * simpleQueue ) backgroundtask ( ) { var retryqueue list . List goglog . Logger . Debug ( " " , t . GetType ( ) ) dur := time . Duration ( t . RetryInterval ) * time . Second ticker := time . NewTicker ( dur ) defer ticker . Stop ( ) for { select { case event := <- t . queue : if ( retryqueue . Len ( ) < t . MaxQueueSize ) || t . MaxQueueSize == - 1 { retryqueue . PushBack ( event ) } case <- t . ctx . Done ( ) : goglog . Logger . Debugf ( " " , t . GetType ( ) ) return case <- ticker . C : if e := retryqueue . Front ( ) ; e != nil { goglog . Logger . Debugf ( " " , t . GetType ( ) , retryqueue . Len ( ) ) if atomic . LoadUint32 ( & t . isInPause ) == StatusPaused { msg := e . Value retryqueue . Remove ( e ) switch v := msg . ( type ) { case logevent . LogEvent : go func ( ) { ctx , cancel := context . WithTimeout ( context . Background ( ) , dur ) err := t . output . OutputEvent ( ctx , v ) if err != nil { goglog . Logger . Errorf ( " " , t . GetType ( ) , err . Error ( ) ) } cancel ( ) } ( ) case [ ] byte : go func ( ) { select { case <- t . ctx . Done ( ) : goglog . Logger . Debugf ( " " , t . GetType ( ) ) return case t . codecCh <- v : } } ( ) default : goglog . Logger . Errorf ( " " , v ) } } else { myList := [ ] interface { } { } for { e := retryqueue . Front ( ) if e == nil { break } event := e . Value myList = append ( myList , event ) retryqueue . Remove ( e ) } go func ( ) { for x := range myList { ctx , cancel := context . WithTimeout ( context . Background ( ) , dur ) defer cancel ( ) switch v := ( myList [ x ] ) . ( type ) { case [ ] byte : select { case <- t . ctx . Done ( ) : return case t . codecCh <- v : } case logevent . LogEvent : err := t . Output ( ctx , v ) if err != nil { goglog . Logger . Errorf ( " " , t . GetType ( ) , err . Error ( ) ) } default : goglog . Logger . Errorf ( " " , v ) } } } ( ) } } } } }
backgroundtask is running in the background and adds new events to the queue, tries to send them out on the RetryInterval.
backgroundtask is running in the background and adds new events to the queue tries to send them out on the RetryInterval .
// list of queued messages; the list is not multithreading safe, and must only be accessed from backgroundtask() // We have reached a RetryInterval. If there are any events in the queue, lets send one back. // If we are still in pause mode we will send one, if we are in normal mode we will send all back. // we are not in pause mode and will queue all the events in the queue for sending. // First we need to empty the queue and get all events to send. // Now we have to send them all out