message

Author: phuhao00

.idea/.gitignore

@@ -31,3 +31,15 @@ func Async[A any](f func() A) chan A {
 
 	return ch
 }
+
+// FanIn ...
+func FanIn(in ...chan any) <-chan any {
+	out := make(chan any)
+	for i := range in {
+		tmp := in[i]
+		go func() {
+			out <- tmp
+		}()
+	}
+	return out
+}

.idea/modules.xml

@@ -0,0 +1,2 @@
+golang.org/x/exp v0.0.0-20220713135740-79cabaa25d75 h1:x03zeu7B2B11ySp+daztnwM5oBJ/8wGUSqrwcw9L0RA=
+golang.org/x/exp v0.0.0-20220713135740-79cabaa25d75/go.mod h1:Kr81I6Kryrl9sr8s2FK3vxD90NdsKWRuOIl2O4CvYbA=

.idea/sugar.iml

@@ -0,0 +1,92 @@
+package sugar
+
+import (
+	"errors"
+	"time"
+)
+
+type Message struct {
+	// Contents
+}
+
+type Subscription struct {
+	ch chan Message
+
+	Inbox chan Message
+}
+
+func (s *Subscription) Publish(msg Message) error {
+	if _, ok := <-s.ch; !ok {
+		return errors.New("Topic has been closed")
+	}
+
+	s.ch <- msg
+
+	return nil
+}
+
+type Topic struct {
+	Subscribers    []Session
+	MessageHistory []Message
+	ch             <-chan Message
+}
+
+func (t *Topic) Subscribe(uid uint64, name string) (Subscription, error) {
+	// Get session and create one if it's the first
+	s := Session{
+		User: User{
+			ID:   uid,
+			Name: name,
+		},
+		Timestamp: time.Now(),
+	}
+
+	// Add session to the Topic & MessageHistory
+	t.Subscribers = append(t.Subscribers, s)
+	// Create a subscription
+	return Subscription{}, nil
+}
+
+func (t *Topic) Unsubscribe(Subscription) error {
+	// Implementation
+	return nil
+}
+
+func (t *Topic) Delete() error {
+	// Implementation
+	return nil
+}
+
+type User struct {
+	ID   uint64
+	Name string
+}
+
+type Session struct {
+	User      User
+	Timestamp time.Time
+}
+
+var (
+	gSubscription Subscription
+	gTopic        Topic
+	c             = make(chan Message)
+)
+
+func work() {
+
+	gSubscription.ch = c
+	gSubscription.Publish(Message{})
+
+}
+
+func hub() {
+	for {
+		select {
+		case data := <-c:
+			gTopic.MessageHistory = append(gTopic.MessageHistory, data)
+
+		}
+	}
+
+}

.idea/vcs.xml

@@ -1 +1,22 @@
 package sugar
+
+type RingBuffer struct {
+	inputChannel  <-chan any
+	outputChannel chan any
+}
+
+func NewRingBuffer(inputChannel <-chan any, outputChannel chan any) *RingBuffer {
+	return &RingBuffer{inputChannel, outputChannel}
+}
+
+func (r *RingBuffer) Run() {
+	for v := range r.inputChannel {
+		select {
+		case r.outputChannel <- v:
+		default:
+			<-r.outputChannel
+			r.outputChannel <- v
+		}
+	}
+	close(r.outputChannel)
+}

concurrency.go

@@ -1 +1,23 @@
 package sugar
+
+import (
+	"fmt"
+	"testing"
+)
+
+func TestRingBuffer(t *testing.T) {
+	in := make(chan any)
+	out := make(chan any, 5)
+	rb := NewRingBuffer(in, out)
+	go rb.Run()
+
+	for i := 0; i < 10; i++ {
+		in <- i
+	}
+
+	close(in)
+
+	for res := range out {
+		fmt.Println(res)
+	}
+}