[Pkg] feat: add channel observable

[bryanchriswhite]

Summary

Summary generated by Reviewpad on 20 Oct 23 08:51 UTC

This pull request contains several file diffs:

1. The diff of the file observer_test.go includes the addition of two test functions, TestObserver_Unsubscribe and TestObserver_ConcurrentUnsubscribe. These test the behavior of the Unsubscribe method in the channelObserver struct.

2. The diff of the file "observer.go" includes the addition of a new package named "channel", several imports, constants, types, functions, and methods related to an implementation of an observer for channels in the "observable" package.

3. The diff for the file README.md includes the addition of comprehensive documentation for the package, including architecture diagrams, installation instructions, features, usage examples, configuration options, API reference, best practices, FAQ, contributing guidelines, changelog, and license information.

4. A new file named "require.go" has been added to the internal/testerrors package. This file contains a package declaration, an import statement, and the definition of a variable named ErrAsync. This variable is used for handling test assertion failures in separate goroutines.

5. The diff adds a new file named "drain.go" to the "internal/testchannel" package. The file contains a function named "DrainChannel" that attempts to receive from a given channel until it is empty. If the channel is not closed by the time it's empty, the function returns an error. Additionally, a TODO_CONSIDERATION comment is included, suggesting the addition of a timeout parameter to the function.

6. The diff shows the addition of a new file README.md in the docs/pkg/observable directory. This file provides comprehensive documentation for the pkg/observable package, including an overview, functionality descriptions, architecture diagrams, usage examples, and considerations for developers.

7. The diff introduces a new file called "errors.go" in the "pkg/observable" directory, which defines a package called "observable" with two global variables and imports another package called "errorsmod" from the "cosmossdk.io/errors" module.

8. A new file called observable.go has been added to the pkg/observable/channel directory. This file implements the observable.Observable interface and provides an implementation for creating and managing observables using channels. It includes functions for creating observables, subscribing and unsubscribing observers, and notifying observers when values are received.

9. The diff in go.mod includes changes related to dependencies, such as adding a new dependency and removing an indirect dependency.

10. The diff introduces a new file called "interface.go" in the "pkg/observable" package, which defines two generic interfaces, Observable[V] and Observer[V], for implementing a notifications package.

11. A new file named observation_test.go has been added in the pkg/observable/channel package. This file contains helper functions and data structures related to observations in the context of channels.

This pull request includes various additions and changes to different files and packages, including tests, documentation, implementation of observers, error handling, and dependency management.

(NOTE: following the standard go project layout guidelines)

Issue

Related to:

• [Miner] Foundation for the Miner submodule #13
• [Miner] feat: add block client #65
• [Miner] feat: add supplier client #42

Observable is a dependency of the ServicerClient, which is a dependency of the Miner.

Type of change

Select one or more:

• New feature, functionality or library
• Bug fix
• Code health or cleanup
• Documentation
• Other (specify)

Testing

• Run all unit tests: make go_test
• Verify Localnet manually: See the instructions [here](TODO: add link to instructions)

Sanity Checklist

• I have tested my changes using the available tooling
• I have performed a self-review of my own code
• I have commented my code, updated documentation and left TODOs throughout the codebase

[Olshansk]

Few things:

1. Great work! This is really difficult work but it's really clear and easy to follow. 🔥

2. The current state of things (other than comments almost non of which are blocking) is g2g so we can hopefully merge it in tomorrow. ✅

3. We need to focus on functionality before tending to the TODOs here, but I am worried that it might come back to bite us, so please make sure to prioritize this afterwards. ⚠️

[Olshansk]

@bryanchriswhite I reviewed all your inline comments and resolved them.

Thanks for clarifying everything with better names & comments. I really think its approachable by anyone now.

A couple small NITs but otherwise LFG!

[Olshansk]

Moving previous convo here: #31 (comment)

Thoughts on doing a if (len(obsvbl.observers) == 0) pain("Should have no observers after unsubscribing")

[bryanchriswhite]

TL;DR, I worry such a change would be more likely to cause some perverse result than to expose any bugs.

I don't think we want to do this. The observers list is explicitly being set to an empty slice at the end of unsubscribeAll(). If we wanted to check that the obervers slice is empty here, we would have to acquire a lock on the mutex. This raises some concerns for me with regards to false positives on the timing; although, this change doesn't seem to break tests. Regardless, this could be refactored by moving such a check directly into unsubscribeAll(). But then you end up with something that looks like this:

// unsubscribeAll unsubscribes and removes all observers from the observable.
func (obsvbl *channelObservable[V]) unsubscribeAll() {
	...

	// Reset observers to an empty list. This purges any observers which might have
	// subscribed while the observable was closing.
	obsvbl.observersMu.Lock()
	obsvbl.observers = []*channelObserver[V]{}
	if (obsvbl.observers) > 0 {
		panic("impossible")
	}
	obsvbl.observersMu.Unlock()
}