proc-listener.go (1950B)
1 // This file is part of taler-cashless2ecash. 2 // Copyright (C) 2024 Joel Häberli 3 // 4 // taler-cashless2ecash is free software: you can redistribute it and/or modify it 5 // under the terms of the GNU Affero General Public License as published 6 // by the Free Software Foundation, either version 3 of the License, 7 // or (at your option) any later version. 8 // 9 // taler-cashless2ecash is distributed in the hope that it will be useful, but 10 // WITHOUT ANY WARRANTY; without even the implied warranty of 11 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 12 // Affero General Public License for more details. 13 // 14 // You should have received a copy of the GNU Affero General Public License 15 // along with this program. If not, see <http://www.gnu.org/licenses/>. 16 // 17 // SPDX-License-Identifier: AGPL3.0-or-later 18 19 package internal_proc 20 21 import ( 22 internal_utils "c2ec/internal/utils" 23 "c2ec/pkg/db" 24 "context" 25 "errors" 26 ) 27 28 func RunListener( 29 ctx context.Context, 30 channel string, 31 callback func(*db.Notification, chan error), 32 notifications chan *db.Notification, 33 errs chan error, 34 ) { 35 36 listenFunc, err := db.DB.NewListener(channel, notifications) 37 if err != nil { 38 internal_utils.LogError("listener", err) 39 errs <- errors.New("failed setting up listener") 40 return 41 } 42 43 go func() { 44 internal_utils.LogInfo("listener", "listener starts listening for notifications at the db for channel="+channel) 45 err := listenFunc(ctx) 46 if err != nil { 47 internal_utils.LogError("listener", err) 48 errs <- err 49 } 50 close(notifications) 51 close(errs) 52 }() 53 54 // Listen is started async. We can therefore block here and must 55 // not run the retrieval logic in own goroutine 56 for { 57 select { 58 case notification := <-notifications: 59 // the dispatching and further processing can be done asynchronously 60 // thus not blocking further incoming notifications. 61 go callback(notification, errs) 62 case <-ctx.Done(): 63 errs <- ctx.Err() 64 return 65 } 66 } 67 }