wait for subscribers to complete

server/subscriber.go

@@ -226,11 +226,15 @@ func (s *rpcServer) createSubHandler(sb *subscriber, opts Options) broker.Handle
 				fn = opts.SubWrappers[i-1](fn)
 			}
 
-			go fn(ctx, &rpcPublication{
+			s.wg.Add(1)
-				topic:       sb.topic,
+			go func() {
-				contentType: ct,
+				fn(ctx, &rpcPublication{
-				message:     req.Interface(),
+					topic:       sb.topic,
-			})
+					contentType: ct,
+					message:     req.Interface(),
+				})
+				s.wg.Done()
+			}()
 		}
 		return nil
 	}