Observer Pattern

1. Definition

• When the state of an object is updated, all objects need to be notified: broadcast.
• There are two roles: subject and observer.
  • The observer observes the subject (listens to the subject).
  • The subject changes.
  • The subject actively notifies the observers.

2. UML

3. Java

3.1. Client

public class Client
{
    public static void main(String[] args)
    {
        ConcreteSubject subject = new ConcreteSubject();

        ObserverA observerA = new ObserverA();
        ObserverA observerA1 = new ObserverA();
        ObserverA observerA2 = new ObserverA();

        subject.addObserver(observerA);
        subject.addObserver(observerA1);
        subject.addObserver(observerA2);

        // Change the subject's state
        subject.setState(3000);
        System.out.println("########################");
        // Check whether the observers' state changes as well
        System.out.println(observerA.getConcreteSubjectState());
        System.out.println(observerA1.getConcreteSubjectState());
        System.out.println(observerA2.getConcreteSubjectState());
    }
}

3.2. Observed Object

public class ConcreteSubject extends Observable
{
    private int state;

    public int getState()
    {
        return state;
    }

    public void setState(int state)
    {
        this.state = state;

        // Automatically call the update method
        setChanged();

        notifyObservers();
    }
}

3.3. Observer

public class ObserverA implements Observer
{
    private int concreteSubjectState;

    @Override
    public void update(Observable o, Object arg)
    {
        this.concreteSubjectState = ((ConcreteSubject)o).getState();
    }

    public int getConcreteSubjectState()
    {
        return concreteSubjectState;
    }

    public void setConcreteSubjectState(int concreteSubjectState)
    {
        this.concreteSubjectState = concreteSubjectState;
    }
}

4. Golang

4.1. Subject

/* Subject */
type ISubject interface {
	Attach(sub ISubscriber)
	UpdateTopic(topic string)
}

type Subject struct {
	subscribers []ISubscriber // Subscribers
	topic       string        // Subject content
}

func NewSubject() *Subject {
	return &Subject{subscribers: make([]ISubscriber, 0)}
}

// Add a subscriber
func (s *Subject) Attach(sub ISubscriber) {
	s.subscribers = append(s.subscribers, sub)
}

// Notify all subscribers
func (s *Subject) nofity() {
	for _, subscriber := range s.subscribers {
		subscriber.Update(s)
	}
}

// Update the subject and notify all subscribers
func (s *Subject) UpdateTopic(topic string) {
	s.topic = topic
	s.nofity()
}

4.2. Subscriber

/* Subscriber */
type ISubscriber interface {
	Update(*Subject)
}

type Subscriber struct {
	name string
}

func NewSubscriber(name string) *Subscriber {
	return &Subscriber{name: name}
}

func (s Subscriber) Update(subject *Subject) {
	fmt.Printf("Topic: [%s] changed, Subscriber: [%s] received
", subject.topic, s.name)
}

4.3. Client

func main() {
	subject := 观察者.NewSubject()
	subject.Attach(观察者.NewSubscriber("subscriber 1"))
	subject.Attach(观察者.NewSubscriber("subscriber 2"))
	subject.UpdateTopic("subject updated")
}

5. Example

• notifier

var GlobalModifiedAnchorNotify = &ModifiedAnchorNotifier{}

type IModifiedAnchorNotifier interface {
	AddObservers(observers ...IObserver)
	NotifyObservers([]*model.AnchorWhitelist)
}

type ModifiedAnchorNotifier struct {
	observers []IObserver
}

func (s *ModifiedAnchorNotifier) AddObservers(observers ...IObserver) {
	s.observers = append(s.observers, observers...)
}

func (s *ModifiedAnchorNotifier) NotifyObservers(ctx context.Context,
	list []*model.AnchorWhitelist) {

	if len(list) == 0 {
		return
	}

	for i := range s.observers {
		s.observers[i].Notify(ctx, list)
	}
}

• observer

type IObserver interface {
	Notify(ctx context.Context, list []*model.AnchorWhitelist)
}

• mini_program_observer

type MiniProgramObserver struct {
}

func (o *MiniProgramObserver) Notify(ctx context.Context, list []*model.AnchorWhitelist) {

	anchorUidsMap := make(map[int32][]uint64)
	for _, anchor := range list {
		anchorUidsMap[anchor.DistributionPlatform] = append(anchorUidsMap[anchor.DistributionPlatform], anchor.Uid)
	}

	for platform, anchorUids := range anchorUidsMap {
		service.[REDACTED_INTERNAL_SERVICE].DeleteMediaId(ctx,
			&pb.DeleteMediaIdReq{
				AnchorUid:    anchorUids,
				PlatformType: platform,
			},
			&pb.DeleteMediaIdRsp{})
	}
}

func init() {
	GlobalModifiedAnchorNotify.AddObservers(&MiniProgramObserver{})
}

• wx_pub_observer

type WxPubObserver struct {
}

func (o *WxPubObserver) Notify(ctx context.Context,
	list []*model.AnchorWhitelist) {

	anchorUidsMap := make(map[int32][]uint64)
	for _, anchor := range list {
		anchorUidsMap[anchor.DistributionPlatform] = append(anchorUidsMap[anchor.DistributionPlatform], anchor.Uid)
	}

	for platform, anchorUids := range anchorUidsMap {
		rpc.DeleteAnchorInfoCache(ctx, anchorUids, platform)
	}
}

func init() {
	GlobalModifiedAnchorNotify.AddObservers(&WxPubObserver{})
}

6. References

• Difference Between Publish–Subscribe and Observer Patterns - SegmentFault