Firebase Notification Service

firebase-notification-service.java

@Service
@Slf4j
public class NotificationService {

    @Autowired
    private Config config;
    private EmitterProcessor<Notification> emitterSource = EmitterProcessor.create();
    private Flux<Notification> fluxNotification = emitterSource.publish().autoConnect();

    @PostConstruct
    private void init() {
        try {
            FirebaseOptions options = new FirebaseOptions.Builder()
                    .setCredentials(GoogleCredentials.fromStream(new ByteArrayInputStream(config.getFirebasePrivateKey().getBytes(StandardCharsets.UTF_8))))
                    .build();
            FirebaseApp.initializeApp(options);
        } catch (IOException e) {
            log.error("Error initializating Firebase SDK");
        }

        //SIMPLE FLUX NOTIFICATION DISPATCHER
        fluxNotification.parallel()
                .doOnNext(this::sendNotification)
                .subscribe();
    }

    //METHOD CALLED BY NOTIFICATON PRODUCER
    public Mono<Void> addNotification(NotifcationTransit notify) {
        emitterSource.onNext(notify);
        return Mono.empty();

    }

    private void sendNotification(NotifcationTransit transit) {
        try {
            FirebaseMessaging.getInstance().send(Message.builder()
                    .setToken(transit.getToken())
                    .setNotification(Notification.builder()
                            .setTitle(NOTIFY_TITLE)
                            .setBody(NOTIFY_BODY)
                            .build())
                    .build());
        } catch (FirebaseMessagingException e) {
            log.error("[{}] {}",e.getErrorCode(),e.getMessage());
        }
    }

}