func (s *TopologyNew) prepareContext() {
	if s.ctx == nil || s.ctx.Err() != nil {
		contextLogger := common.Log.WithField("rule", s.name)
		ctx := contexts.WithValue(contexts.Background(), contexts.LoggerKey, contextLogger)
		s.ctx, s.cancel = ctx.WithCancel()
	}
}
// prepareContext setups internal context before
// stream starts execution.
prepareContext setups internal context before
stream starts execution.
// prepareContext setups internal context before // stream starts execution .
func (s *TopologyNew) prepareContext() {
	if s.ctx == nil || s.ctx.Err() != nil {
		contextLogger := common.Log.WithField("rule", s.name)
		ctx := contexts.WithValue(contexts.Background(), contexts.LoggerKey, contextLogger)
		s.ctx, s.cancel = ctx.WithCancel()
	}
}
func ( s * TopologyNew ) prepareContext ( ) { if s . ctx == nil || s . ctx . Err ( ) != nil { contextLogger := common . Log . WithField ( " " , s . name ) ctx := contexts . WithValue ( contexts . Background ( ) , contexts . LoggerKey , contextLogger ) s . ctx , s . cancel = ctx . WithCancel ( ) } }
prepareContext setups internal context before
stream starts execution.
prepareContext setups internal context before stream starts execution .