rocketmq

package module
v2.2.2 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Feb 4, 2021 License: Apache-2.0 Imports: 5 Imported by: 0

README

RocketMQ Client Go

TravisCI License Go Report Card GoDoc CodeCov GitHub release Average time to resolve an issue Percentage of issues still open Twitter Follow

A product ready RocketMQ Client in pure go, which supports almost the full features of Apache RocketMQ, such as pub and sub messages, ACL, tracing and so on.


Due Diligence

Here, we sincerely invite you to take a minute to feedback on your usage scenario. Click Here or go to ISSUE #423 if you accept.


Features

For 2.0.0 version, it supports:

  • sending message in synchronous mode
  • sending message in asynchronous mode
  • sending message in oneway mode
  • sending orderly messages
  • consuming message using push model
  • message tracing for pub and sub messages
  • ACL for producers and consumers

How to use


Apache RocketMQ Community


Contact us


How to Contribute

Contributions are warmly welcome! Be it trivial cleanup, major new feature or other suggestion. Read this how to contribute guide for more details.


License

Apache License, Version 2.0 Copyright (C) Apache Software Foundation

Documentation

Overview

Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

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

This section is empty.

Variables

View Source
var (
	ErrRequestTimeout = errors.New("request timeout")
	ErrMQEmpty        = errors.New("MessageQueue is nil")
	ErrOffset         = errors.New("offset < 0")
	ErrNumbers        = errors.New("numbers < 0")
)

Functions

This section is empty.

Types

type Producer

type Producer interface {
	Start() error
	Shutdown() error
	SendSync(ctx context.Context, mq ...*primitive.Message) (*primitive.SendResult, error)
	SendAsync(ctx context.Context, mq func(ctx context.Context, result *primitive.SendResult, err error),
		msg ...*primitive.Message) error
	SendOneWay(ctx context.Context, mq ...*primitive.Message) error
}

func NewProducer

func NewProducer(opts ...producer.Option) (Producer, error)

type PullConsumer

type PullConsumer interface {
	// Start the PullConsumer for consuming message
	Start() error

	// Shutdown the PullConsumer, all offset of MessageQueue will be commit to broker before process exit
	Shutdown() error

	// Subscribe a topic for consuming
	Subscribe(topic string, selector consumer.MessageSelector) error

	// Unsubscribe a topic
	Unsubscribe(topic string) error

	// MessageQueues get MessageQueue list about for a given topic. This method will issue a remote call to the server
	// if it does not already have any MessageQueue about the given topic.
	MessageQueues(topic string) []primitive.MessageQueue

	// Pull message for the topic specified. It is an error to not have subscribed to any topics before pull for message
	//
	// Specified numbers of messages is returned if message greater that numbers, and the offset will auto forward.
	// It means that if you meeting messages consuming failed, you should process failed messages by yourself.
	Pull(ctx context.Context, topic string, numbers int) (*primitive.PullResult, error)

	// Pull message for the topic specified from a specified MessageQueue and offset. It is an error to not have
	// subscribed to any topics before pull for message. the method will not affect the offset recorded
	//
	// Specified numbers of messages is returned.
	PullFrom(ctx context.Context, mq primitive.MessageQueue, offset int64, numbers int) (*primitive.PullResult, error)

	// Lookup offset for the given message queue by timestamp. The returned offset for the message queue is the
	// earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding message
	// queue.
	//
	// Timestamp must be millisecond level, if you want to lookup the earliest offset of the mq, you could set the
	// timestamp 0, and if you want to the latest offset the mq, you could set the timestamp math.MaxInt64.
	Lookup(ctx context.Context, mq primitive.MessageQueue, timestamp int64) (int64, error)

	// Commit the offset of specified mqs to broker, if auto-commit is disable, you must commit the offset manually.
	Commit(ctx context.Context, mqs ...primitive.MessageQueue) (int64, error)

	// CommittedOffset return the offset of specified Message
	CommittedOffset(mq primitive.MessageQueue) (int64, error)

	// Seek set offset of the mq, if you wanna re-consuming your message form one position, the method may help you.
	// if you want re-consuming from one time, you cloud Lookup() then seek it.
	Seek(mq primitive.MessageQueue, offset int64) error
}

func NewPullConsumer

func NewPullConsumer(opts ...consumer.Option) (PullConsumer, error)

The PullConsumer has not implemented completely, if you want have an experience of PullConsumer, you could use consumer.NewPullConsumer(...), but it may changed in the future.

The PullConsumer will be supported in next release

type PushConsumer

type PushConsumer interface {
	// Start the PullConsumer for consuming message
	Start() error

	// Shutdown the PullConsumer, all offset of MessageQueue will be sync to broker before process exit
	Shutdown() error
	// Subscribe a topic for consuming
	Subscribe(topic string, selector consumer.MessageSelector,
		f func(context.Context, ...*primitive.MessageExt) (consumer.ConsumeResult, error)) error

	// Unsubscribe a topic
	Unsubscribe(topic string) error
}

func NewPushConsumer

func NewPushConsumer(opts ...consumer.Option) (PushConsumer, error)

type TransactionProducer

type TransactionProducer interface {
	Start() error
	Shutdown() error
	SendMessageInTransaction(ctx context.Context, mq *primitive.Message) (*primitive.TransactionSendResult, error)
}

func NewTransactionProducer

func NewTransactionProducer(listener primitive.TransactionListener, opts ...producer.Option) (TransactionProducer, error)

Directories

Path Synopsis
Package consumer is a generated GoMock package.
Package consumer is a generated GoMock package.
examples
consumer/retry/order
* * use orderly consumer model, when Subscribe function return consumer.SuspendCurrentQueueAMoment, it will be re-send to * local msg queue for later consume if msg.ReconsumeTimes < MaxReconsumeTimes, otherwise, it will be send to rocketmq * DLQ topic, we should manually resolve the msg.
* * use orderly consumer model, when Subscribe function return consumer.SuspendCurrentQueueAMoment, it will be re-send to * local msg queue for later consume if msg.ReconsumeTimes < MaxReconsumeTimes, otherwise, it will be send to rocketmq * DLQ topic, we should manually resolve the msg.
producer/acl
Package main implements a producer with user custom interceptor.
Package main implements a producer with user custom interceptor.
producer/interceptor
Package main implements a producer with user custom interceptor.
Package main implements a producer with user custom interceptor.
producer/namespace
Package main implements a producer with user custom interceptor.
Package main implements a producer with user custom interceptor.
Package internal is a generated GoMock package.
Package internal is a generated GoMock package.
remote
* Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements.
* Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements.
* Define the ctx key and value type.
* Define the ctx key and value type.
* * builtin interceptor
* * builtin interceptor

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL