rocketmq

package module
v2.1.2 Latest Latest
Warning

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

Go to latest
Published: Sep 9, 2022 License: Apache-2.0 Imports: 6 Imported by: 0

README

RocketMQ Client Go

Build Status License Code Scanning Tests CodeCov Go Report Card GoDoc 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.


Features

For 2.X version, it supports:

  • sending message in synchronous mode
  • sending message in asynchronous mode
  • sending message in oneway mode
  • sending message in batch mode
  • sending orderly messages
  • sending delay messages
  • sending transaction messages
  • consuming message using push model
  • consuming message using pull model
  • consuming message using broadcast model
  • message tracing for pub and sub messages
  • ACL for producers and consumers
  • requset-reply model

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

This section is empty.

Functions

This section is empty.

Types

type Producer

type Producer interface {
	GetClient() internal.RMQClient
	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
	Request(ctx context.Context, ttl time.Duration, msg *primitive.Message) (*primitive.Message, error)
	RequestAsync(ctx context.Context, ttl time.Duration, callback internal.RequestCallback, msg *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

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

	// Unsubscribe a topic
	Unsubscribe(topic string) error

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

	// Poll messages with timeout.
	Poll(ctx context.Context, timeout time.Duration) (*consumer.ConsumeRequest, error)

	//ACK ACK
	ACK(ctx context.Context, cr *consumer.ConsumeRequest, consumeResult consumer.ConsumeResult)

	// Pull message of topic,  selector indicate which queue to pull.
	Pull(ctx context.Context, numbers int) (*primitive.PullResult, error)

	// PullFrom pull messages of queue from the offset to offset + numbers
	PullFrom(ctx context.Context, queue *primitive.MessageQueue, offset int64, numbers int) (*primitive.PullResult, error)

	// UpdateOffset updateOffset update offset of queue in mem
	UpdateOffset(queue *primitive.MessageQueue, offset int64) error

	// PersistOffset persist all offset in mem.
	PersistOffset(ctx context.Context, topic string) error

	// CurrentOffset return the current offset of queue in mem.
	CurrentOffset(queue *primitive.MessageQueue) (int64, error)
}

func NewPullConsumer

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

type PushConsumer

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

	// Shutdown the PushConsumer, 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

	// Suspend the consumption
	Suspend()

	// Resume the consumption
	Resume()
}

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
*
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.
*

Jump to

Keyboard shortcuts

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