server

Documentation

Overview

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Index

• Constants
• func CallShell(shellString string) error
• func DoResponse(ctx core.Context, request *protocol.RemotingCommand, ...)
• type BrokerController
  • func NewBrokerController(cfg *config.Config) (*BrokerController, error)
  • func (controller *BrokerController) RegisterConsumeMessageHook(hook trace.ConsumeMessageHook)
  • func (controller *BrokerController) RegisterSendMessageHook(hook trace.SendMessageHook)
  • func (controller *BrokerController) Shutdown()
  • func (controller *BrokerController) Start()
• type ConsumerIdsChangeListener
• type OffsetTable
  • func (table *OffsetTable) Foreach(fn func(k string, v map[int]int64))
  • func (table *OffsetTable) Get(k string) map[int]int64
  • func (j *OffsetTable) MarshalJSON() ([]byte, error)
  • func (j *OffsetTable) MarshalJSONBuf(buf fflib.EncodingBuffer) error
  • func (table *OffsetTable) Put(k string, v map[int]int64)
  • func (table *OffsetTable) PutAll(offsetMap *concurrent.ConcurrentMap)
  • func (table *OffsetTable) Remove(k string) map[int]int64
  • func (table *OffsetTable) RemoveByFlag(fn func(k string, v map[int]int64) bool)
  • func (table *OffsetTable) Size() int
  • func (j *OffsetTable) UnmarshalJSON(input []byte) error
  • func (j *OffsetTable) UnmarshalJSONFFLexer(fs *fflib.FFLexer, state fflib.FFParseState) error
• type SendMessageProcessor
  • func NewSendMessageProcessor(brokerController *BrokerController) *SendMessageProcessor
  • func (smp *SendMessageProcessor) ConsumerSendMsgBack(conn core.Context, request *protocol.RemotingCommand) (remotingCommand *protocol.RemotingCommand)
  • func (smp *SendMessageProcessor) ExecuteConsumeMessageHookAfter(context *trace.ConsumeMessageContext)
  • func (smp *SendMessageProcessor) HasConsumeMessageHook() bool
  • func (smp *SendMessageProcessor) HasSendMessageHook() bool
  • func (smp *SendMessageProcessor) ProcessRequest(ctx core.Context, request *protocol.RemotingCommand) (*protocol.RemotingCommand, error)
  • func (smp *SendMessageProcessor) RegisterConsumeMessageHook(consumeMessageHookList []trace.ConsumeMessageHook)
  • func (smp *SendMessageProcessor) RegisterSendMessageHook(sendMessageHookList []trace.SendMessageHook)
  • func (smp *SendMessageProcessor) SendMessage(ctx core.Context, request *protocol.RemotingCommand, ...) *protocol.RemotingCommand

Constants

const (
	TOPIC_GROUP_SEPARATOR = "@"
	MAX_VALUE             = 0x7fffffffffffffff
)

const (
	DLQ_NUMS_PER_GROUP = 1
)

Functions

func CallShell

func CallShell(shellString string) error

callShell 执行命令 Author rongzhihong Since 2017/9/8

func DoResponse

func DoResponse(ctx core.Context,
	request *protocol.RemotingCommand, response *protocol.RemotingCommand)

Types

type BrokerController

type BrokerController struct {
	// contains filtered or unexported fields
}

BrokerController broker服务控制器 Author gaoyanlei Since 2017/8/25

func NewBrokerController

func NewBrokerController(cfg *config.Config) (*BrokerController, error)

NewBrokerController 创建BrokerController对象

func (*BrokerController) RegisterConsumeMessageHook

func (controller *BrokerController) RegisterConsumeMessageHook(hook trace.ConsumeMessageHook)

RegisterConsumeMessageHook 注册消费消息的回调 Author rongzhihong Since 2017/9/11

func (*BrokerController) RegisterSendMessageHook

func (controller *BrokerController) RegisterSendMessageHook(hook trace.SendMessageHook)

RegisterSendMessageHook 注册发送消息的回调 Author rongzhihong Since 2017/9/11

func (*BrokerController) Shutdown

func (controller *BrokerController) Shutdown()

Shutdown BrokerController停止入口 Author rongzhihong Since 2017/9/12

func (*BrokerController) Start

func (controller *BrokerController) Start()

Start 控制器的start启动入口 Author rongzhihong Since 2017/9/12

type ConsumerIdsChangeListener

type ConsumerIdsChangeListener interface {
	ConsumerIdsChanged(group string, channels []core.Context)
}

type OffsetTable

type OffsetTable struct {
	Offsets map[string]map[int]int64 `json:"offsets"`
	// contains filtered or unexported fields
}

func (*OffsetTable) Foreach

func (table *OffsetTable) Foreach(fn func(k string, v map[int]int64))

func (*OffsetTable) Get

func (table *OffsetTable) Get(k string) map[int]int64

func (*OffsetTable) MarshalJSON

func (j *OffsetTable) MarshalJSON() ([]byte, error)

MarshalJSON marshal bytes to json - template

func (*OffsetTable) MarshalJSONBuf

func (j *OffsetTable) MarshalJSONBuf(buf fflib.EncodingBuffer) error

MarshalJSONBuf marshal buff to json - template

func (*OffsetTable) Put

func (table *OffsetTable) Put(k string, v map[int]int64)

func (*OffsetTable) PutAll

func (table *OffsetTable) PutAll(offsetMap *concurrent.ConcurrentMap)

PutAll 同步Offset配置文件 Author rongzhihong Since 2017/9/18

func (*OffsetTable) Remove

func (table *OffsetTable) Remove(k string) map[int]int64

func (*OffsetTable) RemoveByFlag

func (table *OffsetTable) RemoveByFlag(fn func(k string, v map[int]int64) bool)

func (*OffsetTable) Size

func (table *OffsetTable) Size() int

func (*OffsetTable) UnmarshalJSON

func (j *OffsetTable) UnmarshalJSON(input []byte) error

UnmarshalJSON umarshall json - template of ffjson

func (*OffsetTable) UnmarshalJSONFFLexer

func (j *OffsetTable) UnmarshalJSONFFLexer(fs *fflib.FFLexer, state fflib.FFParseState) error

UnmarshalJSONFFLexer fast json unmarshall - template ffjson

type SendMessageProcessor

type SendMessageProcessor struct {
	// contains filtered or unexported fields
}

SendMessageProcessor 处理客户端发送消息的请求 Author gaoyanlei Since 2017/8/24

func NewSendMessageProcessor

func NewSendMessageProcessor(brokerController *BrokerController) *SendMessageProcessor

func (*SendMessageProcessor) ConsumerSendMsgBack

func (smp *SendMessageProcessor) ConsumerSendMsgBack(conn core.Context,
	request *protocol.RemotingCommand) (remotingCommand *protocol.RemotingCommand)

consumerSendMsgBack 客户端返回未消费消息 Author gaoyanlei Since 2017/8/17

func (*SendMessageProcessor) ExecuteConsumeMessageHookAfter

func (smp *SendMessageProcessor) ExecuteConsumeMessageHookAfter(context *trace.ConsumeMessageContext)

ExecuteConsumeMessageHookAfter 消费消息后执行回调 Author rongzhihong Since 2017/9/5

func (*SendMessageProcessor) HasConsumeMessageHook

func (smp *SendMessageProcessor) HasConsumeMessageHook() bool

HasConsumeMessageHook 判断是否存在消费消息回调 Author rongzhihong Since 2017/9/5

func (*SendMessageProcessor) HasSendMessageHook

func (smp *SendMessageProcessor) HasSendMessageHook() bool

HasSendMessageHook 判断是否存在发送消息回调 Author rongzhihong Since 2017/9/5

func (*SendMessageProcessor) ProcessRequest

func (smp *SendMessageProcessor) ProcessRequest(ctx core.Context, request *protocol.RemotingCommand) (*protocol.RemotingCommand, error)

func (*SendMessageProcessor) RegisterConsumeMessageHook

func (smp *SendMessageProcessor) RegisterConsumeMessageHook(consumeMessageHookList []trace.ConsumeMessageHook)

RegisterSendMessageHook 注册赋值消费消息回调 Author rongzhihong Since 2017/9/5

func (*SendMessageProcessor) RegisterSendMessageHook

func (smp *SendMessageProcessor) RegisterSendMessageHook(sendMessageHookList []trace.SendMessageHook)

RegisterSendMessageHook 注册赋值发送消息回调 Author rongzhihong Since 2017/9/5

func (*SendMessageProcessor) SendMessage

func (smp *SendMessageProcessor) SendMessage(ctx core.Context, request *protocol.RemotingCommand,
	traceContext *trace.SendMessageContext, requestHeader *head.SendMessageRequestHeader) *protocol.RemotingCommand

sendMessage 正常消息 Author gaoyanlei Since 2017/8/17