astra/photoprism-client-go
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Working auth and photo json endpoint

Browse source 
Signed-off-by: Kris Nóva <kris@nivenly.com>
This commit is contained in:
Kris Nóva 2021-02-09 11:17:06 -08:00
parent ef275f97f4
commit e4323b6047
2032 changed files with 821464 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

35
vendor/github.com/leandro-lugaresi/hub/.codecov.yml generated vendored Normal file
View file

@ -0,0 +1,35 @@
codecov:
notify:
require_ci_to_pass: no
coverage:
precision: 2
round: down
range: "70...100"
status:
project:
default:
# basic
target: auto
threshold: 2
base: auto
patch: yes
changes: no
threshold: 2
parsers:
gcov:
branch_detection:
conditional: yes
loop: yes
method: no
macro: no
comment:
layout: "reach, diff, flags, files, footer"
behavior: default
require_changes: no
ignore:
- "**/*_test.go"

34
vendor/github.com/leandro-lugaresi/hub/.gitignore generated vendored Normal file
View file

@ -0,0 +1,34 @@
# Compiled Object files, Static and Dynamic libs (Shared Objects)
*.o
*.a
*.so
*.exe
*.dll
*.prof
debug
# Folders
_obj
_test
dist
*.cgo1.go
*.cgo2.c
_cgo_defun.c
_cgo_gotypes.go
_cgo_export.*
_testmain.go
# Test binary, build with `go test -c`
*.test
# Output of the go coverage tool, specifically when used with LiteIDE
*.out
coverage.txt
# ignored until version 1.0
vendor

88
vendor/github.com/leandro-lugaresi/hub/.golangci.yml generated vendored Normal file
View file

@ -0,0 +1,88 @@
# This file contains all available configuration options
# with their default values.
# options for analysis running
run:
# timeout for analysis, e.g. 30s, 5m, default is 1m
deadline: 10m
# output configuration options
output:
# colored-line-number|line-number|json|tab|checkstyle, default is "colored-line-number"
format: colored-line-number
# print lines of code with issue, default is true
print-issued-lnes: true
# print linter name in the end of issue text, default is true
print-linter-name: true
# all available settings of specific linters
linters-settings:
golint:
# minimal confidence for issues, default is 0.8
min-confidence: 0.8
gocyclo:
# minimal code complexity to report, 30 by default (but we recommend 10-20)
min-complexity: 20
maligned:
# print struct with more effective memory layout or not, false by default
suggest-new: true
goimports:
local-prefixes: github.com/empregoligado/rabbids
depguard:
list-type: blacklist
include-go-root: false
packages:
- github.com/davecgh/go-spew/spew
lll:
# max line length, lines longer will be reported. Default is 120.
# '\t' is counted as 1 character by default, and can be changed with the tab-width option
line-length: 130
# tab width in spaces. Default to 1.
tab-width: 4
unused:
# treat code as a program (not a library) and report unused exported identifiers; default is false.
# XXX: if you enable this setting, unused will report a lot of false-positives in text editors:
# if it's called for subdir of a project it can't find funcs usages. All text editor integrations
# with golangci-lint call it on a directory with the changed file.
check-exported: false
unparam:
# Inspect exported functions, default is false. Set to true if no external program/library imports your code.
# XXX: if you enable this setting, unparam will report a lot of false-positives in text editors:
# if it's called for subdir of a project it can't find external interfaces. All text editor integrations
# with golangci-lint call it on a directory with the changed file.
check-exported: false
wsl:
# If true append is only allowed to be cuddled if appending value is
# matching variables, fields or types on line above. Default is true.
strict-append: true
# Allow calls and assignments to be cuddled as long as the lines have any
# matching variables, fields or types. Default is true.
allow-assign-and-call: true
# Allow multiline assignments to be cuddled. Default is true.
allow-multiline-assign: true
# Allow declarations (var) to be cuddled.
allow-cuddle-declarations: true
# Allow trailing comments in ending of blocks
allow-trailing-comment: false
# Force newlines in end of case at this limit (0 = never).
force-case-trailing-whitespace: 0
linters:
enable-all: true
disable:
- maligned
- prealloc
- gosec
- gochecknoglobals
- goimports
- gomnd
issues:
# List of regexps of issue texts to exclude, empty list by default.
# But independently from this option we use default exclude patterns,
# it can be disabled by `exclude-use-default: false`. To list all
# excluded by default patterns execute `golangci-lint run --help`
# exclude:
# - newHTTP - result 1 (error) is always nil

46
vendor/github.com/leandro-lugaresi/hub/CODE_OF_CONDUCT.md generated vendored Normal file
View file

@ -0,0 +1,46 @@
# Contributor Covenant Code of Conduct
## Our Pledge
In the interest of fostering an open and welcoming environment, we as contributors and maintainers pledge to making participation in our project and our community a harassment-free experience for everyone, regardless of age, body size, disability, ethnicity, gender identity and expression, level of experience, nationality, personal appearance, race, religion, or sexual identity and orientation.
## Our Standards
Examples of behavior that contributes to creating a positive environment include:
* Using welcoming and inclusive language
* Being respectful of differing viewpoints and experiences
* Gracefully accepting constructive criticism
* Focusing on what is best for the community
* Showing empathy towards other community members
Examples of unacceptable behavior by participants include:
* The use of sexualized language or imagery and unwelcome sexual attention or advances
* Trolling, insulting/derogatory comments, and personal or political attacks
* Public or private harassment
* Publishing others' private information, such as a physical or electronic address, without explicit permission
* Other conduct which could reasonably be considered inappropriate in a professional setting
## Our Responsibilities
Project maintainers are responsible for clarifying the standards of acceptable behavior and are expected to take appropriate and fair corrective action in response to any instances of unacceptable behavior.
Project maintainers have the right and responsibility to remove, edit, or reject comments, commits, code, wiki edits, issues, and other contributions that are not aligned to this Code of Conduct, or to ban temporarily or permanently any contributor for other behaviors that they deem inappropriate, threatening, offensive, or harmful.
## Scope
This Code of Conduct applies both within project spaces and in public spaces when an individual is representing the project or its community. Examples of representing a project or community include using an official project e-mail address, posting via an official social media account, or acting as an appointed representative at an online or offline event. Representation of a project may be further defined and clarified by project maintainers.
## Enforcement
Instances of abusive, harassing, or otherwise unacceptable behavior may be reported by contacting the project team at leandrolugaresi92@gmail.com. The project team will review and investigate all complaints, and will respond in a way that it deems appropriate to the circumstances. The project team is obligated to maintain confidentiality with regard to the reporter of an incident. Further details of specific enforcement policies may be posted separately.
Project maintainers who do not follow or enforce the Code of Conduct in good faith may face temporary or permanent repercussions as determined by other members of the project's leadership.
## Attribution
This Code of Conduct is adapted from the [Contributor Covenant][homepage], version 1.4, available at [http://contributor-covenant.org/version/1/4][version]
[homepage]: http://contributor-covenant.org
[version]: http://contributor-covenant.org/version/1/4/

58
vendor/github.com/leandro-lugaresi/hub/CONTRIBUTING.md generated vendored Normal file
View file

@ -0,0 +1,58 @@
# Contributing
By participating to this project, you agree to abide our [code of
conduct](/CODE_OF_CONDUCT.md).
## Setup your machine
`hub` is written in [Go](https://golang.org/).
Prerequisites are:
- Build:
- `make`
- [Go 1.9+](http://golang.org/doc/install)
Clone `hub` from source into some path:
```sh
git clone git@github.com:leandro-lugaresi/hub.git
cd hub
```
If you created a fork clone your fork and add my repository as a upstream remote:
```sh
git clone git@github.com:{your-name}/hub.git
cd hub
git remote add upstream git@github.com:leandro-lugaresi/hub.git
```
### Install
Install the build and lint dependencies:
```sh
$ make setup
```
A good way of making sure everything is all right is running the test suite:
```sh
$ make test
```
## Test your change
When you are satisfied with the changes, we suggest you run:
```sh
$ make ci
```
Which runs all the linters and tests.
## Submit a pull request
Push your branch to your `example` fork and open a pull request against the
main branch.

201
vendor/github.com/leandro-lugaresi/hub/LICENSE generated vendored Normal file
View file

@ -0,0 +1,201 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [2018] [Leandro Lugaresi]
Licensed 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.

35
vendor/github.com/leandro-lugaresi/hub/Makefile generated vendored Normal file
View file

@ -0,0 +1,35 @@
SOURCE_FILES?=$$(go list ./... | grep -v /vendor/)
TEST_PATTERN?=./...
TEST_OPTIONS?=-race
setup: ## Install all the build and lint dependencies
sudo curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $$(go env GOPATH)/bin v1.22.2
GO111MODULE=off go get github.com/mfridman/tparse
GO111MODULE=off go get -u golang.org/x/tools/cmd/cover
go get -v -t ./...
test: ## Run all the tests
go test $(TEST_OPTIONS) -covermode=atomic -coverprofile=coverage.txt -timeout=1m -cover -json $(SOURCE_FILES) | $$(go env GOPATH)/bin/tparse -all
throughput: ## Run the throughput tests
go test -v -timeout 60s github.com/leandro-lugaresi/hub -run ^TestThroughput -args -throughput
bench: ## Run the benchmark tests
go test -bench=. $(TEST_PATTERN)
cover: test ## Run all the tests and opens the coverage report
go tool cover -html=coverage.txt
fmt: ## gofmt and goimports all go files
find . -name '*.go' -not -wholename './vendor/*' | while read -r file; do gofmt -w -s "$$file"; goimports -w "$$file"; done
lint: ## Run all the linters
$$(go env GOPATH)/bin/golangci-lint run
ci: lint test ## Run all the tests and code checks
# Absolutely awesome: http://marmelab.com/blog/2016/02/29/auto-documented-makefile.html
help:
@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}'
.DEFAULT_GOAL := help

146
vendor/github.com/leandro-lugaresi/hub/README.md generated vendored Normal file
View file

@ -0,0 +1,146 @@
# Hub
:incoming_envelope: A fast enough Event Hub for go applications using publish/subscribe with support patterns on topics like rabbitMQ exchanges.
[![Release](https://img.shields.io/github/release/leandro-lugaresi/hub.svg?style=flat-square)](https://github.com/leandro-lugaresi/hub/releases/latest)
[![Software License](https://img.shields.io/github/license/leandro-lugaresi/hub.svg?style=flat-square)](LICENSE.md)
[![Actions Status](https://github.com/leandro-lugaresi/hub/workflows/Go/badge.svg)](https://github.com/leandro-lugaresi/hub/actions)
[![Coverage Status](https://img.shields.io/codecov/c/github/leandro-lugaresi/hub/main.svg?style=flat-square)](https://codecov.io/gh/leandro-lugaresi/hub)
[![Go Doc](https://img.shields.io/badge/godoc-reference-blue.svg?style=flat-square)](http://godoc.org/github.com/leandro-lugaresi/hub)
[![Go Report Card](https://goreportcard.com/badge/github.com/leandro-lugaresi/hub?style=flat-square)](https://goreportcard.com/report/github.com/leandro-lugaresi/hub)
[![Say Thanks!](https://img.shields.io/badge/Say%20Thanks-!-1EAEDB.svg)](https://saythanks.io/to/leandro-lugaresi)
---
## Table of Contents
- [Install](#install)
- [Usage](#usage)
- [Examples & Demos](#examples--demos)
- [Benchmarks](#benchmarks)
- [CSTrie](#cstrie)
- [Contribute](CONTRIBUTING.md)
- [Code of conduct](CODE_OF_CONDUCT.md)
## Install
To install this library you can `go get` it but I encourage you to always vendor your dependencies or use one of the version tags of this project.
```sh
go get -u github.com/leandro-lugaresi/hub
```
```sh
dep ensure --add github.com/leandro-lugaresi/hub
```
## Usage
### Subscribers
Hub provides subscribers as buffered (cap > `0`) and unbuffered (cap = 0) channels but we have two different types of subscribers:
- `Subscriber` this is the default subscriber and it's a blocking subscriber so if the channel is full and you try to send another message the send operation will block until the subscriber consumes some message.
- `NonBlockingSubscriber` this subscriber will never block on the publish side but if the capacity of the channel is reached the publish operation will be lost and an alert will be trigged.
This should be used only if loose data is acceptable. ie: metrics, logs
### Topics
This library uses the same concept of topic exchanges on rabbiMQ, so the message name is used to find all the subscribers that match the topic, like a route.
The topic must be a list of words delimited by dots (`.`) however, there is one important special case for binding keys:
`*` (star) can substitute for exactly one word.
## Examples & Demos
```go
package main
import (
"fmt"
"sync"
"github.com/leandro-lugaresi/hub"
)
func main() {
h := hub.New()
var wg sync.WaitGroup
// the cap param is used to create one buffered channel with cap = 10
// If you wan an unbuferred channel use the 0 cap
sub := h.Subscribe(10, "account.login.*", "account.changepassword.*")
wg.Add(1)
go func(s hub.Subscription) {
for msg := range s.Receiver {
fmt.Printf("receive msg with topic %s and id %d\n", msg.Name, msg.Fields["id"])
}
wg.Done()
}(sub)
h.Publish(hub.Message{
Name: "account.login.failed",
Fields: hub.Fields{"id": 123},
})
h.Publish(hub.Message{
Name: "account.changepassword.failed",
Fields: hub.Fields{"id": 456},
})
h.Publish(hub.Message{
Name: "account.login.success",
Fields: hub.Fields{"id": 123},
})
// message not routed to this subscriber
h.Publish(hub.Message{
Name: "account.foo.failed",
Fields: hub.Fields{"id": 789},
})
// close all the subscribers
h.Close()
// wait until finish all the messages on buffer
wg.Wait()
// Output:
// receive msg with topic account.login.failed and id 123
// receive msg with topic account.changepassword.failed and id 456
// receive msg with topic account.login.success and id 123
}
```
See more [here](https://godoc.org/github.com/leandro-lugaresi/hub#example-Hub)!
## Benchmarks
To run the benchmarks you can execute:
```bash
make bench
```
Currently, I only have the benchmarks of the CSTrie used internally. I will provide more benchmarks.
## Throughput
The project have one test for throughput, just execute:
```bash
make throughput
```
In a intel(R) core(TM) i5-4460 CPU @ 3.20GHz x4 we got this results:
```
go test -v -timeout 60s github.com/leandro-lugaresi/hub -run ^TestThroughput -args -throughput
=== RUN TestThroughput
1317530.091292 msg/sec
--- PASS: TestThroughput (3.04s)
PASS
ok github.com/leandro-lugaresi/hub 3.192s
```
## CSTrie
This project uses internally an awesome Concurrent Subscription Trie done by [@tylertreat](https://github.com/tylertreat). If you want to learn more about see this [blog post](http://bravenewgeek.com/fast-topic-matching/) and the code is [here](https://github.com/tylertreat/fast-topic-matching)
---
This project adheres to the Contributor Covenant [code of conduct](CODE_OF_CONDUCT.md). By participating, you are expected to uphold this code.
We appreciate your contribution. Please refer to our [contributing guidelines](CONTRIBUTING.md) for further information

5
vendor/github.com/leandro-lugaresi/hub/go.mod generated vendored Normal file
View file

@ -0,0 +1,5 @@
module github.com/leandro-lugaresi/hub
go 1.13
require github.com/stretchr/testify v1.4.0

10
vendor/github.com/leandro-lugaresi/hub/go.sum generated vendored Normal file
View file

@ -0,0 +1,10 @@
github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

100
vendor/github.com/leandro-lugaresi/hub/hub.go generated vendored Normal file
View file

@ -0,0 +1,100 @@
package hub
// AlertTopic is used to notify when a nonblocking subscriber loose one message
// You can subscribe on this topic and log or send metrics.
const AlertTopic = "hub.subscription.messageslost"
type (
//Hub is a component that provides publish and subscribe capabilities for messages.
// Every message has a Name used to route them to subscribers and this can be used like RabbitMQ topics exchanges.
// Where every word is separated by dots `.` and you can use `*` as a wildcard.
Hub struct {
matcher matcher
fields Fields
}
)
// New create and return a new empty hub.
func New() *Hub {
return &Hub{
matcher: newCSTrieMatcher(),
fields: Fields{},
}
}
// Publish will send an event to all the subscribers matching the event name.
func (h *Hub) Publish(m Message) {
for k, v := range h.fields {
m.Fields[k] = v
}
for _, sub := range h.matcher.Lookup(m.Topic()) {
sub.Set(m)
}
}
// With creates a child Hub with the fields added to it.
// When someone call Publish, this Fields will be added automatically into the message.
func (h *Hub) With(f Fields) *Hub {
hub := Hub{
matcher: h.matcher,
fields: Fields{},
}
for k, v := range h.fields {
hub.fields[k] = v
}
for k, v := range f {
hub.fields[k] = v
}
return &hub
}
// Subscribe create a blocking subscription to receive events for a given topic.
// The cap param is used inside the subscriber and in this case used to create a channel.
// cap(1) = unbuffered channel.
func (h *Hub) Subscribe(cap int, topics ...string) Subscription {
return h.matcher.Subscribe(topics, newBlockingSubscriber(cap))
}
// NonBlockingSubscribe create a nonblocking subscription to receive events for a given topic.
// This subscriber will loose messages if the buffer reaches the max capability.
func (h *Hub) NonBlockingSubscribe(cap int, topics ...string) Subscription {
return h.matcher.Subscribe(
topics,
newNonBlockingSubscriber(
cap,
alertFunc(func(missed int) {
h.alert(missed, topics)
}),
))
}
// Unsubscribe remove and close the Subscription.
func (h *Hub) Unsubscribe(sub Subscription) {
h.matcher.Unsubscribe(sub)
sub.subscriber.Close()
}
// Close will unsubscribe all the subscriptions and close them all.
func (h *Hub) Close() {
subs := h.matcher.Subscriptions()
for _, s := range subs {
h.matcher.Unsubscribe(s)
}
for _, s := range subs {
s.subscriber.Close()
}
}
func (h *Hub) alert(missed int, topics []string) {
h.Publish(Message{
Name: AlertTopic,
Fields: Fields{
"missed": missed,
"topic": topics,
},
})
}

58
vendor/github.com/leandro-lugaresi/hub/matching.go generated vendored Normal file
View file

@ -0,0 +1,58 @@
// Copyright (C) 2018 Tyler Treat <https://github.com/tylertreat>
// Licensed 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.
// Modifications copyright (C) 2018 Leandro Lugaresi
package hub
const (
delimiter = "."
wildcard = "*"
)
type (
// Subscription represents a topic subscription.
Subscription struct {
Topics []string
Receiver <-chan Message
subscriber subscriber
}
// subscriber is the interface used internally to send values and get the channel used by subscribers.
// This is used to override the behaviour of channel and support nonBlocking operations
subscriber interface {
// Set send the given Event to be processed by the subscriber
Set(Message)
// Ch return the channel used to consume messages inside the subscription.
// This func MUST always return the same channel.
Ch() <-chan Message
// Close will close the internal state and the subscriber will not receive more messages
// WARN: This function can be executed more than one time so the code MUST take care of this situation and
// avoid problems like close a closed channel.
Close()
}
)
// matcher contains topic subscriptions and performs matches on them.
type matcher interface {
// Subscribe adds the Subscriber to the topics and returns a Subscription.
Subscribe(topics []string, sub subscriber) Subscription
// Unsubscribe removes the Subscription.
Unsubscribe(sub Subscription)
// Lookup returns the subscribers for the given topic.
Lookup(topic string) []subscriber
Subscriptions() []Subscription
}

610
vendor/github.com/leandro-lugaresi/hub/matching_cstrie.go generated vendored Normal file
View file

@ -0,0 +1,610 @@
// Copyright (C) 2018 Tyler Treat <https://github.com/tylertreat>
// Licensed 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.
// Modifications copyright (C) 2018 Leandro Lugaresi
package hub
import (
"strings"
"sync/atomic"
"unsafe"
)
type iNode struct {
main *mainNode
}
type mainNode struct {
cNode *cNode
tNode *tNode
}
type cNode struct {
branches map[string]*branch
}
// newCNode creates a new C-node with the given subscription path.
func newCNode(words []string, sub subscriber) *cNode {
if len(words) == 1 {
return &cNode{
branches: map[string]*branch{
words[0]: {subs: map[subscriber]struct{}{sub: {}}},
},
}
}
nin := &iNode{main: &mainNode{cNode: newCNode(words[1:], sub)}}
return &cNode{
branches: map[string]*branch{
words[0]: {subs: map[subscriber]struct{}{}, iNode: nin},
},
}
}
// inserted returns a copy of this C-node with the specified subscriber
// inserted.
func (c *cNode) inserted(words []string, sub subscriber) *cNode {
branches := make(map[string]*branch, len(c.branches)+1)
for key, branch := range c.branches {
branches[key] = branch
}
var br *branch
if len(words) == 1 {
br = &branch{subs: map[subscriber]struct{}{sub: {}}}
} else {
br = &branch{
subs: make(map[subscriber]struct{}),
iNode: &iNode{main: &mainNode{cNode: newCNode(words[1:], sub)}},
}
}
branches[words[0]] = br
return &cNode{branches: branches}
}
// updated returns a copy of this C-node with the specified branch updated.
func (c *cNode) updated(word string, sub subscriber) *cNode {
branches := make(map[string]*branch, len(c.branches))
for word, branch := range c.branches {
branches[word] = branch
}
newBranch := &branch{subs: map[subscriber]struct{}{sub: {}}}
br, ok := branches[word]
if ok {
for id, sub := range br.subs {
newBranch.subs[id] = sub
}
newBranch.iNode = br.iNode
}
branches[word] = newBranch
return &cNode{branches: branches}
}
// updatedBranch returns a copy of this C-node with the specified branch
// updated.
func (c *cNode) updatedBranch(word string, in *iNode, br *branch) *cNode {
branches := make(map[string]*branch, len(c.branches))
for key, branch := range c.branches {
branches[key] = branch
}
branches[word] = br.updated(in)
return &cNode{branches: branches}
}
// removed returns a copy of this C-node with the subscriber removed from the
// corresponding branch.
func (c *cNode) removed(word string, sub subscriber) *cNode {
branches := make(map[string]*branch, len(c.branches))
for word, branch := range c.branches {
branches[word] = branch
}
br, ok := branches[word]
if ok {
br = br.removed(sub)
if len(br.subs) == 0 && br.iNode == nil {
// Remove the branch if it contains no subscribers and doesn't
// point anywhere.
delete(branches, word)
} else {
branches[word] = br
}
}
return &cNode{branches: branches}
}
// getBranches returns the branches for the given word. There are two possible
// branches: exact match and single wildcard.
func (c *cNode) getBranches(word string) (*branch, *branch) {
return c.branches[word], c.branches[wildcard]
}
type branch struct {
iNode *iNode
subs map[subscriber]struct{}
}
// updated returns a copy of this branch updated with the given I-node.
func (b *branch) updated(in *iNode) *branch {
subs := make(map[subscriber]struct{}, len(b.subs))
for id, sub := range b.subs {
subs[id] = sub
}
return &branch{subs: subs, iNode: in}
}
// removed returns a copy of this branch with the given subscriber removed.
func (b *branch) removed(sub subscriber) *branch {
subs := make(map[subscriber]struct{}, len(b.subs))
for id, sub := range b.subs {
subs[id] = sub
}
delete(subs, sub)
return &branch{subs: subs, iNode: b.iNode}
}
// subscribers returns the Subscribers for this branch.
func (b *branch) subscribers() []subscriber {
subs := make([]subscriber, len(b.subs))
i := 0
for sub := range b.subs {
subs[i] = sub
i++
}
return subs
}
type tNode struct{}
type csTrieMatcher struct {
root *iNode
}
func newCSTrieMatcher() matcher {
root := &iNode{main: &mainNode{cNode: &cNode{}}}
return &csTrieMatcher{root: root}
}
// Subscribe adds the subscriber to the topic and returns a Subscription.
func (c *csTrieMatcher) Subscribe(topics []string, sub subscriber) Subscription {
var (
rootPtr = (*unsafe.Pointer)(unsafe.Pointer(&c.root))
root = (*iNode)(atomic.LoadPointer(rootPtr))
)
for _, topic := range topics {
words := strings.Split(topic, delimiter)
if !c.iinsert(root, nil, words, sub) {
return c.Subscribe(topics, sub)
}
}
return Subscription{Topics: topics, Receiver: sub.Ch(), subscriber: sub}
}
func (c *csTrieMatcher) iinsert(i, parent *iNode, words []string, sub subscriber) bool {
// Linearization point.
mainPtr := (*unsafe.Pointer)(unsafe.Pointer(&i.main))
main := (*mainNode)(atomic.LoadPointer(mainPtr))
switch {
case main.cNode != nil:
cn := main.cNode
br := cn.branches[words[0]]
if br == nil {
// If the relevant branch is not in the map, a copy of the C-node
// with the new entry is created. The linearization point is a
// successful CAS.
ncn := &mainNode{cNode: cn.inserted(words, sub)}
return atomic.CompareAndSwapPointer(
mainPtr, unsafe.Pointer(main), unsafe.Pointer(ncn))
}
// If the relevant key is present in the map, its corresponding
// branch is read.
if len(words) > 1 {
// If more than 1 word is present in the path, the tree must be
// traversed deeper.
if br.iNode != nil {
// If the branch has an I-node, iinsert is called
// recursively.
return c.iinsert(br.iNode, i, words[1:], sub)
}
// Otherwise, an I-node which points to a new C-node must be
// added. The linearization point is a successful CAS.
nin := &iNode{main: &mainNode{cNode: newCNode(words[1:], sub)}}
ncn := &mainNode{cNode: cn.updatedBranch(words[0], nin, br)}
return atomic.CompareAndSwapPointer(
mainPtr, unsafe.Pointer(main), unsafe.Pointer(ncn))
}
if _, ok := br.subs[sub]; ok {
// Already subscribed.
return true
}
// Insert the subscriber by copying the C-node and updating the
// respective branch. The linearization point is a successful CAS.
ncn := &mainNode{cNode: cn.updated(words[0], sub)}
return atomic.CompareAndSwapPointer(mainPtr, unsafe.Pointer(main), unsafe.Pointer(ncn))
case main.tNode != nil:
clean(parent)
return false
default:
panic("csTrie is in an invalid state")
}
}
// Unsubscribe removes the Subscription.
func (c *csTrieMatcher) Unsubscribe(sub Subscription) {
var (
rootPtr = (*unsafe.Pointer)(unsafe.Pointer(&c.root))
root = (*iNode)(atomic.LoadPointer(rootPtr))
)
for _, topic := range sub.Topics {
words := strings.Split(topic, delimiter)
if !c.iremove(root, nil, nil, words, 0, sub.subscriber) {
c.Unsubscribe(sub)
}
}
}
func (c *csTrieMatcher) iremove(i, parent, parentsParent *iNode, words []string, wordIdx int, sub subscriber) bool {
// Linearization point.
mainPtr := (*unsafe.Pointer)(unsafe.Pointer(&i.main))
main := (*mainNode)(atomic.LoadPointer(mainPtr))
switch {
case main.cNode != nil:
cn := main.cNode
br := cn.branches[words[wordIdx]]
if br == nil {
// If the relevant word is not in the map, the subscription doesn't
// exist.
return true
}
// If the relevant word is present in the map, its corresponding
// branch is read.
if wordIdx+1 < len(words) {
// If more than 1 word is present in the path, the tree must be
// traversed deeper.
if br.iNode != nil {
// If the branch has an I-node, iremove is called
// recursively.
return c.iremove(br.iNode, i, parent, words, wordIdx+1, sub)
}
// Otherwise, the subscription doesn't exist.
return true
}
if _, ok := br.subs[sub]; !ok {
// Not subscribed.
return true
}
// Remove the subscriber by copying the C-node without it. A
// contraction of the copy is then created. A successful CAS will
// substitute the old C-node with the copied C-node, thus removing
// the subscriber from the trie - this is the linearization point.
ncn := cn.removed(words[wordIdx], sub)
cntr := c.toContracted(ncn, i)
if atomic.CompareAndSwapPointer(
mainPtr, unsafe.Pointer(main), unsafe.Pointer(cntr)) {
if parent != nil {
mainPtr = (*unsafe.Pointer)(unsafe.Pointer(&i.main))
main = (*mainNode)(atomic.LoadPointer(mainPtr))
if main.tNode != nil {
cleanParent(i, parent, parentsParent, c, words[wordIdx-1])
}
}
return true
}
return false
case main.tNode != nil:
clean(parent)
return false
default:
panic("csTrie is in an invalid state")
}
}
// Lookup returns the Subscribers for the given topic.
func (c *csTrieMatcher) Lookup(topic string) []subscriber {
var (
words = strings.Split(topic, delimiter)
rootPtr = (*unsafe.Pointer)(unsafe.Pointer(&c.root))
root = (*iNode)(atomic.LoadPointer(rootPtr))
)
result, ok := c.ilookup(root, nil, words)
if !ok {
return c.Lookup(topic)
}
return result
}
// ilookup attempts to retrieve the Subscribers for the word path. True is
// returned if the Subscribers were retrieved, false if the operation needs to
// be retried.
func (c *csTrieMatcher) ilookup(i, parent *iNode, words []string) ([]subscriber, bool) {
// Linearization point.
mainPtr := (*unsafe.Pointer)(unsafe.Pointer(&i.main))
main := (*mainNode)(atomic.LoadPointer(mainPtr))
switch {
case main.cNode != nil:
// Traverse exact-match branch and single-word-wildcard branch.
exact, singleWC := main.cNode.getBranches(words[0])
subs := make(map[subscriber]struct{})
if exact != nil {
s, ok := c.bLookup(i, exact, words)
if !ok {
return nil, false
}
for _, sub := range s {
subs[sub] = struct{}{}
}
}
if singleWC != nil {
s, ok := c.bLookup(i, singleWC, words)
if !ok {
return nil, false
}
for _, sub := range s {
subs[sub] = struct{}{}
}
}
s := make([]subscriber, len(subs))
i := 0
for sub := range subs {
s[i] = sub
i++
}
return s, true
case main.tNode != nil:
clean(parent)
return nil, false
default:
panic("csTrie is in an invalid state")
}
}
// bLookup attempts to retrieve the Subscribers from the word path along the
// given branch. True is returned if the Subscribers were retrieved, false if
// the operation needs to be retried.
func (c *csTrieMatcher) bLookup(i *iNode, b *branch, words []string) ([]subscriber, bool) {
if len(words) > 1 {
// If more than 1 key is present in the path, the tree must be
// traversed deeper.
if b.iNode == nil {
// If the branch doesn't point to an I-node, no subscribers
// exist.
return make([]subscriber, 0), true
}
// If the branch has an I-node, ilookup is called recursively.
return c.ilookup(b.iNode, i, words[1:])
}
// Retrieve the subscribers from the branch.
return b.subscribers(), true
}
// Subscriptions return all the subscriptions inside the cstrie.
func (c *csTrieMatcher) Subscriptions() []Subscription {
var (
rootPtr = (*unsafe.Pointer)(unsafe.Pointer(&c.root))
root = (*iNode)(atomic.LoadPointer(rootPtr))
)
result, ok := c.isubscriptions(root, nil, []string{})
if !ok {
return c.Subscriptions()
}
return result
}
func (c *csTrieMatcher) isubscriptions(i, parent *iNode, words []string) ([]Subscription, bool) {
// Linearization point.
mainPtr := (*unsafe.Pointer)(unsafe.Pointer(&i.main))
main := (*mainNode)(atomic.LoadPointer(mainPtr))
subs := []Subscription{}
switch {
case main.cNode != nil:
// Traverse all branches.
for word, br := range main.cNode.branches {
cwords := append([]string{}, words...)
cwords = append(cwords, word)
if br.iNode != nil {
// If the branch has an I-node, isubscriptions is called recursively.
s, ok := c.isubscriptions(br.iNode, i, cwords)
if !ok {
return nil, false
}
subs = append(subs, s...)
}
for s := range br.subs {
subs = append(subs, Subscription{
Topics: []string{strings.Join(cwords, delimiter)},
subscriber: s,
Receiver: s.Ch(),
})
}
}
return subs, true
case main.tNode != nil:
clean(parent)
return nil, false
default:
panic("csTrie is in an invalid state")
}
}
// toContracted ensures that every I-node except the root points to a C-node
// with at least one branch or a T-node. If a given C-node has no branches and
// is not at the root level, a T-node is returned.
func (c *csTrieMatcher) toContracted(cn *cNode, parent *iNode) *mainNode {
if c.root != parent && len(cn.branches) == 0 {
return &mainNode{tNode: &tNode{}}
}
return &mainNode{cNode: cn}
}
// clean replaces an I-node's C-node with a copy that has any tombed I-nodes
// resurrected.
func clean(i *iNode) {
mainPtr := (*unsafe.Pointer)(unsafe.Pointer(&i.main))
main := (*mainNode)(atomic.LoadPointer(mainPtr))
if main.cNode != nil {
atomic.CompareAndSwapPointer(mainPtr,
unsafe.Pointer(main), unsafe.Pointer(toCompressed(main.cNode)))
}
}
// cleanParent reads the main node of the parent I-node p and the current
// I-node i and checks if the T-node below i is reachable from p. If i is no
// longer reachable, some other thread has already completed the contraction.
// If it is reachable, the C-node below p is replaced with its contraction.
func cleanParent(i, parent, parentsParent *iNode, c *csTrieMatcher, word string) {
var (
mainPtr = (*unsafe.Pointer)(unsafe.Pointer(&i.main))
main = (*mainNode)(atomic.LoadPointer(mainPtr))
pMainPtr = (*unsafe.Pointer)(unsafe.Pointer(&parent.main))
pMain = (*mainNode)(atomic.LoadPointer(pMainPtr))
)
if pMain.cNode != nil {
if br, ok := pMain.cNode.branches[word]; ok {
if br.iNode != i {
return
}
if main.tNode != nil {
if !contract(parentsParent, parent, c, pMain) {
cleanParent(parentsParent, parent, i, c, word)
}
}
}
}
}
// contract performs a contraction of the parent's C-node if possible. Returns
// true if the contraction succeeded, false if it needs to be retried.
func contract(parentsParent, parent *iNode, c *csTrieMatcher, pMain *mainNode) bool {
ncn := toCompressed(pMain.cNode)
if len(ncn.cNode.branches) == 0 && parentsParent != nil {
// If the compressed C-node has no branches, it and the I-node above it
// should be removed. To do this, a CAS must occur on the parent I-node
// of the parent to update the respective branch of the C-node below it
// to point to nil.
ppMainPtr := (*unsafe.Pointer)(unsafe.Pointer(&parentsParent.main))
ppMain := (*mainNode)(atomic.LoadPointer(ppMainPtr))
for pKey, pBranch := range ppMain.cNode.branches {
// Find the branch pointing to the parent.
if pBranch.iNode == parent {
// Update the branch to point to nil.
updated := ppMain.cNode.updatedBranch(pKey, nil, pBranch)
if len(pBranch.subs) == 0 {
// If the branch has no subscribers, simply prune it.
delete(updated.branches, pKey)
}
// Replace the main node of the parent's parent.
return atomic.CompareAndSwapPointer(ppMainPtr,
unsafe.Pointer(ppMain), unsafe.Pointer(toCompressed(updated)))
}
}
} else {
// Otherwise, perform a simple contraction to a T-node.
cntr := c.toContracted(ncn.cNode, parent)
pMainPtr := (*unsafe.Pointer)(unsafe.Pointer(&parent.main))
pMain := (*mainNode)(atomic.LoadPointer(pMainPtr))
if !atomic.CompareAndSwapPointer(pMainPtr, unsafe.Pointer(pMain),
unsafe.Pointer(cntr)) {
return false
}
}
return true
}
// toCompressed prunes any branches to tombed I-nodes and returns the
// compressed main node.
func toCompressed(cn *cNode) *mainNode {
branches := make(map[string]*branch, len(cn.branches))
for key, br := range cn.branches {
if !prunable(br) {
branches[key] = br
}
}
return &mainNode{cNode: &cNode{branches: branches}}
}
// prunable indicates if the branch can be pruned. A branch can be pruned if
// it has no subscribers and points to nowhere or it has no subscribers and
// points to a tombed I-node.
func prunable(br *branch) bool {
if len(br.subs) > 0 {
return false
}
if br.iNode == nil {
return true
}
mainPtr := (*unsafe.Pointer)(unsafe.Pointer(&br.iNode.main))
main := (*mainNode)(atomic.LoadPointer(mainPtr))
return main.tNode != nil
}

40
vendor/github.com/leandro-lugaresi/hub/message.go generated vendored Normal file
View file

@ -0,0 +1,40 @@
package hub
import (
"fmt"
"sort"
"strings"
)
type (
// Fields is a [key]value storage for Messages values.
Fields map[string]interface{}
// Message represent some message/event passed into the hub
// It also contain some helper functions to convert the fields to primitive types.
Message struct {
Name string
Body []byte
Fields Fields
}
)
// Topic return the message topic used when the message was sended.
func (m *Message) Topic() string {
return m.Name
}
func (f Fields) String() string {
if len(f) == 0 {
return "Fields(<empty>)"
}
fields := make([]string, 0, len(f))
for k, v := range f {
fields = append(fields, fmt.Sprintf("[%s]%v", k, v))
}
sort.Strings(fields)
return "Fields( " + strings.Join(fields, ", ") + " )"
}

83
vendor/github.com/leandro-lugaresi/hub/subscriber.go generated vendored Normal file
View file

@ -0,0 +1,83 @@
package hub
import (
"sync"
)
type (
alertFunc func(missed int)
nonBlockingSubscriber struct {
ch chan Message
alert alertFunc
onceClose sync.Once
}
// blockingSubscriber uses an channel to receive events.
blockingSubscriber struct {
ch chan Message
onceClose sync.Once
}
)
// newNonBlockingSubscriber returns a new nonBlockingSubscriber
// this subscriber will never block when sending an message, if the capacity is full
// we will ignore the message and call the Alert function from the Alerter.
func newNonBlockingSubscriber(cap int, alerter alertFunc) *nonBlockingSubscriber {
if cap <= 0 {
cap = 10
}
return &nonBlockingSubscriber{
ch: make(chan Message, cap),
alert: alerter,
}
}
// Set inserts the given Event into the diode.
func (s *nonBlockingSubscriber) Set(msg Message) {
select {
case s.ch <- msg:
default:
s.alert(1)
}
}
// Ch return the channel used by subscriptions to consume messages.
func (s *nonBlockingSubscriber) Ch() <-chan Message {
return s.ch
}
// Close will close the internal channel and stop receiving messages.
func (s *nonBlockingSubscriber) Close() {
s.onceClose.Do(func() {
close(s.ch)
})
}
// newBlockingSubscriber returns a blocking subscriber using chanels imternally.
func newBlockingSubscriber(cap int) *blockingSubscriber {
if cap < 0 {
cap = 0
}
return &blockingSubscriber{
ch: make(chan Message, cap),
}
}
// Set will send the message using the channel.
func (s *blockingSubscriber) Set(msg Message) {
s.ch <- msg
}
// Ch return the channel used by subscriptions to consume messages.
func (s *blockingSubscriber) Ch() <-chan Message {
return s.ch
}
// Close will close the internal channel and stop receiving messages.
func (s *blockingSubscriber) Close() {
s.onceClose.Do(func() {
close(s.ch)
})
}