add update listener

README.md

@@ -107,15 +107,17 @@ func main() {
 ### Receive updates
 
 ```go
-responses := make(chan client.Type, 100)
-tdlibClient, err := client.NewClient(authorizer, client.WithListener(responses))
+tdlibClient, err := client.NewClient(authorizer)
 if err != nil {
     log.Fatalf("NewClient error: %s", err)
 }
 
-for response := range responses {
-    if response.GetClass() == client.ClassUpdate {
-        log.Printf("%#v", response)
+listener := tdlibClient.GetListener()
+defer listener.Close()
+ 
+for update := range listener.Updates {
+    if update.GetClass() == client.ClassUpdate {
+        log.Printf("%#v", update)
     }
 }
 ```

client/client.go

@@ -10,7 +10,7 @@ type Client struct {
     jsonClient     *JsonClient
     extraGenerator ExtraGenerator
     catcher        chan *Response
-    listeners      []chan Type
+    listenerStore  *listenerStore
     catchersStore  *sync.Map
 }
 
@@ -22,19 +22,13 @@ func WithExtraGenerator(extraGenerator ExtraGenerator) Option {
     }
 }
 
-func WithListener(listener chan Type) Option {
-    return func(client *Client) {
-        client.listeners = append(client.listeners, listener)
-    }
-}
-
 func NewClient(authorizationStateHandler AuthorizationStateHandler, options ...Option) (*Client, error) {
     catchersListener := make(chan *Response, 1000)
 
     client := &Client{
         jsonClient:    NewJsonClient(),
         catcher:       catchersListener,
-        listeners:     []chan Type{},
+        listenerStore: newListenerStore(),
         catchersStore: &sync.Map{},
     }
 
@@ -70,8 +64,16 @@ func (client *Client) receive() {
             continue
         }
 
-        for _, listener := range client.listeners {
-            listener <- typ
+        needGc := false
+        for _, listener := range client.listenerStore.Listeners() {
+            if listener.IsActive() {
+                listener.Updates <- typ
+            } else {
+                needGc = true
+            }
+        }
+        if needGc {
+            client.listenerStore.gc()
         }
     }
 }
@@ -109,3 +111,13 @@ func (client *Client) Send(req Request) (*Response, error) {
         return nil, errors.New("timeout")
     }
 }
+
+func (client *Client) GetListener() *Listener {
+    listener := &Listener{
+        isActive: true,
+        Updates:  make(chan Type, 1000),
+    }
+    client.listenerStore.Add(listener)
+
+    return listener
+}

client/listener.go

@@ -0,0 +1,66 @@
+package client
+
+import (
+    "sync"
+)
+
+func newListenerStore() *listenerStore {
+    return &listenerStore{
+        listeners: []*Listener{},
+    }
+}
+
+type listenerStore struct {
+    sync.Mutex
+    listeners []*Listener
+}
+
+func (store *listenerStore) Add(listener *Listener) {
+    store.Lock()
+    defer store.Unlock()
+
+    store.listeners = append(store.listeners, listener)
+}
+
+func (store *listenerStore) Listeners() []*Listener {
+    store.Lock()
+    defer store.Unlock()
+
+    return store.listeners
+}
+
+func (store *listenerStore) gc() {
+    store.Lock()
+    defer store.Unlock()
+
+    oldListeners := store.listeners
+
+    store.listeners = []*Listener{}
+
+    for _, listener := range oldListeners {
+        if listener.IsActive() {
+            store.listeners = append(store.listeners, listener)
+        }
+    }
+}
+
+type Listener struct {
+    mu       sync.Mutex
+    isActive bool
+    Updates  chan Type
+}
+
+func (listener *Listener) Close() {
+    listener.mu.Lock()
+    defer listener.mu.Unlock()
+
+    listener.isActive = false
+    close(listener.Updates)
+}
+
+func (listener *Listener) IsActive() bool {
+    listener.mu.Lock()
+    defer listener.mu.Unlock()
+
+    return listener.isActive
+}