> uploadtext_

package main

import (
	"fmt"
	"sync"
	"time"
)

// Event represents a generic event
type Event interface{}

// EventBus manages event subscriptions and publishing
type EventBus struct {
	subscribers map[string][]chan Event
	mu          sync.RWMutex
}

// NewEventBus creates a new EventBus instance
func NewEventBus() *EventBus {
	return &EventBus{
		subscribers: make(map[string][]chan Event),
	}
}

// Publish sends an event to all subscribers of a specific topic
func (eb *EventBus) Publish(topic string, event Event) {
	eb.mu.RLock()
	defer eb.mu.RUnlock()

	if channels, found := eb.subscribers[topic]; found {
		for _, ch := range channels {
			// Using non-blocking send to avoid blocking the publisher if a subscriber is slow
			select {
			case ch <- event:
			default:
				fmt.Printf("Warning: Subscriber for topic '%s' not ready, dropping event: %v
", topic, event)
			}
		}
	}
}

// Subscribe adds a channel to listen for events on a specific topic
// Returns the subscriber channel and a function to unsubscribe.
func (eb *EventBus) Subscribe(topic string) (<-chan Event, func()) {
	ch := make(chan Event, 10) // Buffered channel for subscriber
	eb.mu.Lock()
	eb.subscribers[topic] = append(eb.subscribers[topic], ch)
	eb.mu.Unlock()

	unsubscribe := func() {
		eb.mu.Lock()
		defer eb.mu.Unlock()
		if channels, found := eb.subscribers[topic]; found {
			for i, subscriberCh := range channels {
				if subscriberCh == ch {
					// Remove the channel
					eb.subscribers[topic] = append(channels[:i], channels[i+1:]...)
					close(ch) // Close the subscriber's channel
					return
				}
			}
		}
	}
	return ch, unsubscribe
}

func main() {
	bus := NewEventBus()
	var wg sync.WaitGroup

	// Subscriber 1 for "user_events"
	userSub1, unsubscribeUser1 := bus.Subscribe("user_events")
	wg.Add(1)
	go func() {
		defer wg.Done()
		for event := range userSub1 {
			fmt.Printf("User Subscriber 1 received: %v
", event)
		}
		fmt.Println("User Subscriber 1 stopped.")
	}()

	// Subscriber 2 for "order_events"
	orderSub1, unsubscribeOrder1 := bus.Subscribe("order_events")
	wg.Add(1)
	go func() {
		defer wg.Done()
		for event := range orderSub1 {
			fmt.Printf("Order Subscriber 1 received: %v
", event)
		}
		fmt.Println("Order Subscriber 1 stopped.")
	}()

	// Subscriber 3 for both "user_events" and "order_events"
	userSub2, unsubscribeUser2 := bus.Subscribe("user_events")
	wg.Add(1)
	go func() {
		defer wg.Done()
		for event := range userSub2 {
			fmt.Printf("User Subscriber 2 received: %v
", event)
		}
		fmt.Println("User Subscriber 2 stopped.
")	}()

	orderSub2, unsubscribeOrder2 := bus.Subscribe("order_events")
	wg.Add(1)
	go func() {
		defer wg.Done()
		for event := range orderSub2 {
			fmt.Printf("Order Subscriber 2 received: %v
", event)
		}
		fmt.Println("Order Subscriber 2 stopped.")
	}()

	time.Sleep(10 * time.Millisecond) // Give subscribers time to start

	// Publish events
	bus.Publish("user_events", "UserRegistered: Alice")
	bus.Publish("order_events", "OrderCreated: #12345")
	bus.Publish("user_events", "UserLoggedIn: Bob")
	bus.Publish("system_events", "SystemRestarted") // No subscribers for this topic

	time.Sleep(50 * time.Millisecond) // Give time for events to be processed

	// Unsubscribe one listener
	unsubscribeUser1()
	fmt.Println("Unsubscribed User Subscriber 1.")

	bus.Publish("user_events", "UserLoggedOut: Alice") // Only UserSub2 should get this
	bus.Publish("order_events", "OrderShipped: #12345")

	time.Sleep(50 * time.Millisecond)

	// Unsubscribe all
	unsubscribeUser2()
	unsubscribeOrder1()
	unsubscribeOrder2()
	fmt.Println("All subscribers unsubscribed.")

	wg.Wait()
	fmt.Println("Main finished.")
}