Question about project without any subscribers about cap HOT 21 CLOSED

fail: DotNetCore.CAP.Internal.DefaultSubscriberExecutor[5]
      Consumer method 'Group:cap.default.group, Topic:CapDemo' failed to execute.
System.ArgumentOutOfRangeException: Index was out of range. Must be non-negative and less than the size of the collection.
Parameter name: index
   at System.ThrowHelper.ThrowArgumentOutOfRange_IndexException()
   at System.Collections.Generic.List`1.get_Item(Int32 index)
   at DotNetCore.CAP.Internal.DefaultSubscriberExecutor.<ExecuteAsync>d__7.MoveNext() in G:\codes\CAP\src\DotNetCore.CAP\Internal\ISubscriberExecutor.Default.cs:line 45
warn: DotNetCore.CAP.SubscribeQueueExecutor[2]
      Job failed to execute. Will retry.
System.ArgumentOutOfRangeException: Index was out of range. Must be non-negative and less than the size of the collection.
Parameter name: index
   at System.ThrowHelper.ThrowArgumentOutOfRange_IndexException()
   at System.Collections.Generic.List`1.get_Item(Int32 index)
   at DotNetCore.CAP.Internal.DefaultSubscriberExecutor.<ExecuteAsync>d__7.MoveNext() in G:\codes\CAP\src\DotNetCore.CAP\Internal\ISubscriberExecutor.Default.cs:line 45

  capPublisher.Publish("", inserted_person, sqlTrans);

        [CapSubscribe(name: "", Group = RabbitMQOptions.DefaultExchangeName)]
        public void ReceiveMessage(Person person)
        {

        }

namespace DemoA
{
    public class Startup
    {
        public Startup(IConfiguration configuration)
        {
            Configuration = configuration;
        }

        public IConfiguration Configuration { get; }

        public void ConfigureServices(IServiceCollection services)
        {
            services.AddCap(x =>
            {
                x.UseRabbitMQ(opt =>
                {
                    opt.HostName = "192.168.0.89";
                    opt.UserName = "admin";
                    opt.Password = "admin";
                });
                x.UseSqlServer("Data Source=192.168.0.250;Initial Catalog=CapDemo;User ID=sa;Password=123123;");
            });

            services.AddMvc();
        }

        public void Configure(IApplicationBuilder app, IHostingEnvironment env)
        {
            if (env.IsDevelopment())
            {
                app.UseDeveloperExceptionPage();
            }
            app.UseCap();
            app.UseMvc();
        }
    }
}

namespace DemoB
{
    public class Startup
    {
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddCap(x=> {
                x.UseRabbitMQ(opt=> {
                    opt.HostName = "192.168.0.89";
                    opt.UserName = "admin";
                    opt.Password = "admin";
                });
                x.UseSqlServer("Data Source=192.168.0.250;Initial Catalog=cap_trans_test;User ID=sa;Password=123123;");
            });
            services.AddMvc();
        }

        public void Configure(IApplicationBuilder app, IHostingEnvironment env)
        {
            if (env.IsDevelopment())
            {
                app.UseDeveloperExceptionPage();
            }

            app.UseMvc();
            app.UseCap();
        }
    }
}

namespace DemoA.Controllers
{
    [Route("api/demoa/[controller]")]
    public class PersonController : Controller
    {
        private ICapPublisher capPublisher;

        public PersonController(ICapPublisher _publisher)
        {
            this.capPublisher = _publisher;
        }

        [HttpPost]
        public JsonResult Post()
        {
            var sqlOptions = (SqlServerOptions)HttpContext.RequestServices.GetService(typeof(SqlServerOptions));
            var cstr = sqlOptions.ConnectionString;
            using (var sqlconnection = new SqlConnection(cstr))
            {
                sqlconnection.Open();
                using (var sqlTrans = sqlconnection.BeginTransaction())
                {
                    try
                    {
                        List<Person> persons = new List<Person>();
                        for (int i = 0; i < 10000; i++)
                        {
                            var person = new Person { Name = $"Name_{i.ToString()}", Age = i };
                            var inserted_person = PersonInsert(sqlTrans, person);
                            capPublisher.Publish("", inserted_person, sqlTrans);
                            persons.Add(inserted_person);
                            Console.WriteLine(inserted_person);
                        }
                        sqlTrans.Commit();
                        return Json(persons);
                    }
                    catch (Exception ex)
                    {
                        sqlTrans.Rollback();
                        return Json(ex.Message);
                    }
                }
            }
        }

        /// <summary>
        /// 本地业务逻辑
        /// </summary>
        /// <param name="trans"></param>
        /// <param name="person"></param>
        /// <returns></returns>
        [NonAction]
        private Person PersonInsert(SqlTransaction trans, Person person)
        {
            var sqlconnection = trans.Connection;
            var sql = $"insert into person (Name, Age) output inserted.id, inserted.name, inserted.age values ('{person.Name}', {person.Age})";
            var result = sqlconnection.QueryFirstOrDefault<Person>(sql: sql, transaction: trans);
            return result;
        }

        //没有用到
        [NonAction]
        [CapSubscribe("RemovePerson")]
        public void RemovePerson(Person person)
        {
            var cstr = "Data Source=192.168.0.250;Initial Catalog=CapDemo;User ID=sa;Password=123123;";
            using (var sqlconnection = new SqlConnection(cstr))
            {
                sqlconnection.Open();
                using (var trans = sqlconnection.BeginTransaction())
                {
                    try
                    {
                        int row_count = sqlconnection.Execute($"delete from Person where id={person.id}", null, trans);
                        if (row_count == 1)
                        {
                            Console.WriteLine("删除记录成功, {0}", person.ToString());
                        }
                        else
                            Console.WriteLine("删除记录失败, {0}", person.ToString());
                        trans.Commit();
                    }
                    catch (Exception ex)
                    {
                        Console.WriteLine(ex.Message);
                        trans.Rollback();
                    }
                }
            }
        }
    }
}

namespace DemoB.Controers
{
    [Route("api/demob/[controller]")]
    public class GradeController : Controller
    {
        [NonAction]
        [CapSubscribe(name: "", Group = RabbitMQOptions.DefaultExchangeName)]
        public void ReceiveMessage(Person person)
        {
            Console.WriteLine("this.HttpContext =>  {0}", this.HttpContext);

            Console.WriteLine("Grade service 接收:" + person);
            var cstr = "Data Source=192.168.0.250;Initial Catalog=cap_trans_test;User ID=sa;Password=123123;";

            using (var sqlconnection = new SqlConnection(cstr))
            {
                sqlconnection.Open();
                using (var trans = sqlconnection.BeginTransaction())
                {
                    try
                    {
                        var grade = new Grade { personId = person.id, name = "新增班级" };
                        var inserted = sqlconnection.QueryFirstOrDefault<Grade>("insert into Grade(personId,Name) output inserted.id, inserted.personId,inserted.Name values(@personId,@Name)", grade, trans);
                        Console.WriteLine("Grade Inserted: {0}", grade);
                        trans.Commit();
                    }
                    catch (Exception ex)
                    {
                        Console.WriteLine(ex.Message);
                        trans.Rollback();
                    }
                }
            }
        }
    }
}

  capPublisher.Publish("CapPerson", inserted_person, sqlTrans);

        [NonAction]
        [CapSubscribe(name: "CapPerson")]
        public void ReceiveMessage(Person person){…}

var executeDescriptorGroup = _selector.GetTopicExector(receivedMessage.Name); //这里的Name是CapPerson
if (!executeDescriptorGroup.ContainsKey(receivedMessage.Group))// Group是cap.default.group
 {
         var error = $"Topic:{receivedMessage.Name}, can not be found subscriber method.";
         throw new SubscriberNotFoundException(error);
}

// If there are multiple consumers in the same group, we will take the first
var executeDescriptor = executeDescriptorGroup[receivedMessage.Group][0]; //这里没有获取到

// ...

 public IEnumerable<string> Topics { get; private set; }
...
        public void Subscribe(IEnumerable<string> topics)
        {
            if (topics == null) throw new ArgumentNullException(nameof(topics));

            this.Topics = topics; //赋值

            foreach (var topic in topics)
                _channel.QueueBind(_queueName, _exchageName, topic);
        }

        private void RegisterMessageProcessor(IConsumerClient client)
        {
            client.OnMessageReceived += (sender, message) =>
            {
                _logger.EnqueuingReceivedMessage(message.Name, message.Content);

                //这里重新入队
                if (!client.Topics.Contains(message.Name))
                {
                    client.Reject();
                }
                else
                {
                    using (var scope = _serviceProvider.CreateScope())
                    {
                        try
                        {
                            StoreMessage(scope, message);
                            client.Commit();
                        }
                        catch (Exception e)
                        {
                            _logger.LogError(e, "An exception occurred when storage received message. Message:'{0}'.", message);
                            client.Reject();
                        }
                    }
                }
                Pulse();
            };

            client.OnError += (sender, reason) => { _logger.MessageQueueError(reason); };
        }