| 71 | } |
| 72 | |
| 73 | int PartitionChannelBase::Init(int num_partition_kinds, |
| 74 | PartitionParser* partition_parser, |
| 75 | const char* load_balancer_name, |
| 76 | const PartitionChannelOptions* options_in) { |
| 77 | if (num_partition_kinds <= 0) { |
| 78 | LOG(ERROR) << "Parameter[num_partition_kinds] must be positive"; |
| 79 | return -1; |
| 80 | } |
| 81 | if (NULL == partition_parser) { |
| 82 | LOG(ERROR) << "Parameter[partition_parser] must be non-NULL"; |
| 83 | return -1; |
| 84 | } |
| 85 | PartitionChannelOptions options; |
| 86 | if (options_in) { |
| 87 | options = *options_in; |
| 88 | } |
| 89 | options.succeed_without_server = true; |
| 90 | options.log_succeed_without_server = false; |
| 91 | _subs = new (std::nothrow) SubChannel[num_partition_kinds]; |
| 92 | if (NULL == _subs) { |
| 93 | LOG(ERROR) << "Fail to new Channels[" << num_partition_kinds << "]"; |
| 94 | return -1; |
| 95 | } |
| 96 | for (int i = 0; i < num_partition_kinds; ++i) { |
| 97 | if (_subs[i].Init("list://", load_balancer_name, &options) != 0) { |
| 98 | LOG(ERROR) << "Fail to init sub channel[" << i << "]"; |
| 99 | return -1; |
| 100 | } |
| 101 | } |
| 102 | for (int i = 0; i < num_partition_kinds; ++i) { |
| 103 | if (AddChannel(&_subs[i], DOESNT_OWN_CHANNEL, |
| 104 | options.call_mapper.get(), |
| 105 | options.response_merger.get()) != 0) { |
| 106 | LOG(ERROR) << "Fail to add sub channel[" << i << "]"; |
| 107 | return -1; |
| 108 | } |
| 109 | } |
| 110 | ParallelChannelOptions pchan_options; |
| 111 | pchan_options.timeout_ms = options.timeout_ms; |
| 112 | pchan_options.fail_limit = options.fail_limit; |
| 113 | if (ParallelChannel::Init(&pchan_options) != 0) { |
| 114 | LOG(ERROR) << "Fail to init PartitionChannel as ParallelChannel"; |
| 115 | return -1; |
| 116 | } |
| 117 | // Must be last one because it's the marker of initialized(). |
| 118 | _parser = partition_parser; |
| 119 | return 0; |
| 120 | } |
| 121 | |
| 122 | void PartitionChannelBase::PartitionServersIntoTemps( |
| 123 | const std::vector<ServerId>& servers) { |
no test coverage detected