code: func (s *DataMsgServiceServer) DataChanges(stream msg.DataMsgService_DataChangesServer) error {
  	for {
  		chng, err := stream.Recv()
  
  		if err == io.EOF {
  			return nil
  		}
  		if err != nil {
  			return err
  		}
  
  		for _, sub := range s.adapter.base.Subscriptions() {
  			for _, keyPrefix := range sub.KeyPrefixes {
  				if strings.HasPrefix(chng.Key, keyPrefix) {
  					sub.ChangeChan <- msg.NewChangeWatchResp(context.Background(), chng, func(err2 error) {
  						err = stream.Send(&msg.DataChangeReply{Key: chng.Key, OperationType: chng.OperationType,
  							Result: 0 /*TODO VPP Result*/})
  						if err != nil {
  							logrus.DefaultLogger().Error(err) //Not able to propagate it somewhere else
  						}
  					})
  				}
  			}
  		}
  	}
  }
docs: // DataChanges propagates the events in the stream to go channels of registered plugins.