import org.glassfish.jersey.media.sse.InboundEvent; //导入依赖的package包/类
@Override
public void onEvent(InboundEvent inboundEvent) {
	String eventName = inboundEvent.getName();
	if (mainEventName.equals(eventName)) {
		Optional<Response<T>> resp;
		try {
			resp = responseBuilder.buildResponse(inboundEvent.readData(), clazz, uri);
		} catch (Exception e) {
			sseObserver.onParseError(new ParseError(inboundEvent.getId(), e, inboundEvent.getRawData()));
			return;
		}
		sseObserver.onEvent(resp.get());
	} else {
		sseObserver.onSignal(eventName);
	}

}
 

import org.glassfish.jersey.media.sse.InboundEvent; //导入依赖的package包/类
@Test
public void testFetch() throws Exception {
    final CountDownLatch latch = new CountDownLatch(10);
    final EventSource source = client.fetch();

    final EventListener listener = new EventListener() {
        @Override
        public void onEvent(InboundEvent inboundEvent) {
            latch.countDown();
        }
    };
    source.register(listener, "message-to-client");
    source.open();
    latch.await(10, TimeUnit.SECONDS);
    source.close();
}
 

import org.glassfish.jersey.media.sse.InboundEvent; //导入依赖的package包/类
@Ignore
@Test
public void testSse() {
    Client client = ClientBuilder.newBuilder()
            .register(SseFeature.class).build();
    WebTarget target = client.target("http://localhost:8080/restskol/api")
            .path("events");

    EventInput eventInput = target.request().get(EventInput.class);
    while (!eventInput.isClosed()) {
        final InboundEvent inboundEvent = eventInput.read();
        if (inboundEvent == null) {
            break;
        }

        System.out.println(inboundEvent.getName() + "; "
                + inboundEvent.readData(String.class));
    }
}
 

import org.glassfish.jersey.media.sse.InboundEvent; //导入依赖的package包/类
/**
 * Allows to stream SSE events from horizon.
 * Certain endpoints in Horizon can be called in streaming mode using Server-Sent Events.
 * This mode will keep the connection to horizon open and horizon will continue to return
 * responses as ledgers close.
 * @see <a href="http://www.w3.org/TR/eventsource/" target="_blank">Server-Sent Events</a>
 * @see <a href="https://www.stellar.org/developers/horizon/learn/responses.html" target="_blank">Response Format documentation</a>
 * @param listener {@link EventListener} implementation with {@link AccountResponse} type
 * @return EventSource object, so you can <code>close()</code> connection when not needed anymore
 */
public EventSource stream(final EventListener<AccountResponse> listener) {
  Client client = ClientBuilder.newBuilder().register(SseFeature.class).build();
  WebTarget target = client.target(this.buildUri());
  EventSource eventSource = new EventSource(target) {
    @Override
    public void onEvent(InboundEvent inboundEvent) {
      String data = inboundEvent.readData(String.class);
      if (data.equals("\"hello\"")) {
        return;
      }
      AccountResponse account = GsonSingleton.getInstance().fromJson(data, AccountResponse.class);
      listener.onEvent(account);
    }
  };
  return eventSource;
}
 

import org.glassfish.jersey.media.sse.InboundEvent; //导入依赖的package包/类
/**
 * Allows to stream SSE events from horizon.
 * Certain endpoints in Horizon can be called in streaming mode using Server-Sent Events.
 * This mode will keep the connection to horizon open and horizon will continue to return
 * responses as ledgers close.
 * @see <a href="http://www.w3.org/TR/eventsource/" target="_blank">Server-Sent Events</a>
 * @see <a href="https://www.stellar.org/developers/horizon/learn/responses.html" target="_blank">Response Format documentation</a>
 * @param listener {@link EventListener} implementation with {@link OperationResponse} type
 * @return EventSource object, so you can <code>close()</code> connection when not needed anymore
 */
public EventSource stream(final EventListener<OperationResponse> listener) {
  Client client = ClientBuilder.newBuilder().register(SseFeature.class).build();
  WebTarget target = client.target(this.buildUri());
  EventSource eventSource = new EventSource(target) {
    @Override
    public void onEvent(InboundEvent inboundEvent) {
      String data = inboundEvent.readData(String.class);
      if (data.equals("\"hello\"")) {
        return;
      }
      OperationResponse payment = GsonSingleton.getInstance().fromJson(data, OperationResponse.class);
      listener.onEvent(payment);
    }
  };
  return eventSource;
}
 

import org.glassfish.jersey.media.sse.InboundEvent; //导入依赖的package包/类
/**
 * Allows to stream SSE events from horizon.
 * Certain endpoints in Horizon can be called in streaming mode using Server-Sent Events.
 * This mode will keep the connection to horizon open and horizon will continue to return
 * responses as ledgers close.
 * @see <a href="http://www.w3.org/TR/eventsource/" target="_blank">Server-Sent Events</a>
 * @see <a href="https://www.stellar.org/developers/horizon/learn/responses.html" target="_blank">Response Format documentation</a>
 * @param listener {@link EventListener} implementation with {@link EffectResponse} type
 * @return EventSource object, so you can <code>close()</code> connection when not needed anymore
 */
public EventSource stream(final EventListener<EffectResponse> listener) {
  Client client = ClientBuilder.newBuilder().register(SseFeature.class).build();
  WebTarget target = client.target(this.buildUri());
  EventSource eventSource = new EventSource(target) {
    @Override
    public void onEvent(InboundEvent inboundEvent) {
      String data = inboundEvent.readData(String.class);
      if (data.equals("\"hello\"")) {
        return;
      }
      EffectResponse effect = GsonSingleton.getInstance().fromJson(data, EffectResponse.class);
      listener.onEvent(effect);
    }
  };
  return eventSource;
}
 

import org.glassfish.jersey.media.sse.InboundEvent; //导入依赖的package包/类
/**
 * Allows to stream SSE events from horizon.
 * Certain endpoints in Horizon can be called in streaming mode using Server-Sent Events.
 * This mode will keep the connection to horizon open and horizon will continue to return
 * responses as ledgers close.
 * @see <a href="http://www.w3.org/TR/eventsource/" target="_blank">Server-Sent Events</a>
 * @see <a href="https://www.stellar.org/developers/horizon/learn/responses.html" target="_blank">Response Format documentation</a>
 * @param listener {@link EventListener} implementation with {@link TransactionResponse} type
 * @return EventSource object, so you can <code>close()</code> connection when not needed anymore
 */
public EventSource stream(final EventListener<TransactionResponse> listener) {
  Client client = ClientBuilder.newBuilder().register(SseFeature.class).build();
  WebTarget target = client.target(this.buildUri());
  EventSource eventSource = new EventSource(target) {
    @Override
    public void onEvent(InboundEvent inboundEvent) {
      String data = inboundEvent.readData(String.class);
      if (data.equals("\"hello\"")) {
        return;
      }
      TransactionResponse transaction = GsonSingleton.getInstance().fromJson(data, TransactionResponse.class);
      listener.onEvent(transaction);
    }
  };
  return eventSource;
}
 

import org.glassfish.jersey.media.sse.InboundEvent; //导入依赖的package包/类
/**
 * Allows to stream SSE events from horizon.
 * Certain endpoints in Horizon can be called in streaming mode using Server-Sent Events.
 * This mode will keep the connection to horizon open and horizon will continue to return
 * responses as ledgers close.
 * @see <a href="http://www.w3.org/TR/eventsource/" target="_blank">Server-Sent Events</a>
 * @see <a href="https://www.stellar.org/developers/horizon/learn/responses.html" target="_blank">Response Format documentation</a>
 * @param listener {@link EventListener} implementation with {@link LedgerResponse} type
 * @return EventSource object, so you can <code>close()</code> connection when not needed anymore
 */
public EventSource stream(final EventListener<LedgerResponse> listener) {
  Client client = ClientBuilder.newBuilder().register(SseFeature.class).build();
  WebTarget target = client.target(this.buildUri());
  EventSource eventSource = new EventSource(target) {
    @Override
    public void onEvent(InboundEvent inboundEvent) {
      String data = inboundEvent.readData(String.class);
      if (data.equals("\"hello\"")) {
        return;
      }
      LedgerResponse ledger = GsonSingleton.getInstance().fromJson(data, LedgerResponse.class);
      listener.onEvent(ledger);
    }
  };
  return eventSource;
}
 

import org.glassfish.jersey.media.sse.InboundEvent; //导入依赖的package包/类
@Test
public void test() throws Exception {
    Client client = ClientBuilder.newBuilder().register(SseFeature.class).build();
    WebTarget target = client.target(uri("/events"));

    List<InboundEvent> events = new ArrayList<>();
    CountDownLatch latch = new CountDownLatch(3);

    EventSource eventSource = EventSource.target(target).build();
    EventListener listener = inboundEvent -> {
        events.add(inboundEvent);
        latch.countDown();
    };
    eventSource.register(listener);
    eventSource.open();

    sse.send(event("Hello 1\nLine 1").withId("1").withName("hello1"));
    sse.send(event("Hello 2\nLine 2").withId("2").withName("hello2"));
    sse.send(event("Hello 3\nLine 3").withId("3").withName("hello3"));

    assertTrue(latch.await(500, TimeUnit.MILLISECONDS));

    assertThat(events.get(1).getId(), equalTo("2"));
    assertThat(events.get(1).getName(), equalTo("hello2"));
    assertThat(events.get(1).readData(), equalTo("Hello 2\nLine 2"));
}
 

import org.glassfish.jersey.media.sse.InboundEvent; //导入依赖的package包/类
public void subscribeToNewMessageEvents(Consumer<Message> callback) {
    Thread thread = new Thread() {
        @Override
        public void run() {
            stopListening = false;
            EventInput eventInput = getWebTarget().path("status")
                    .request()
                    .header(HttpHeaders.AUTHORIZATION, getAuthorizationHeader())
                    .get(EventInput.class);
            while (!eventInput.isClosed() && !stopListening) {
                final InboundEvent inboundEvent = eventInput.read();
                if (inboundEvent == null) {
                    // connection has been closed
                    break;
                }
                if ("new-message".equals(inboundEvent.getName())) {
                    Message message = inboundEvent.readData(Message.class);
                    if (message != null) {
                        callback.accept(message);
                    }
                }
            }
        }
    };
    thread.setDaemon(true);
    thread.start();
}
 

import org.glassfish.jersey.media.sse.InboundEvent; //导入依赖的package包/类
@Test
public void testSendDataWithValidAuthentication() throws InterruptedException {
    //given
    final ServerSentEventService serverSentEventService = Application.getInstance(ServerSentEventService.class);
    final Config config = Application.getInstance(Config.class);
    final String data = "Server sent data with authentication FTW!";

    //when
    final WebTarget target = ClientBuilder.newBuilder()
            .register(SseFeature.class)
            .build()
            .target("http://" + config.getConnectorHttpHost() + ":" + config.getConnectorHttpPort() + "/sseauth");

    final CustomWebTarget customWebTarget = new CustomWebTarget(target, new Cookie(COOKIE_NAME, VALID_COOKIE_VALUE));
    final EventSource eventSource = EventSource.target(customWebTarget).build();
    final EventListener listener = new EventListener() {
        @Override
        public void onEvent(InboundEvent inboundEvent) {
            if (StringUtils.isBlank(eventData)) {
                eventData = inboundEvent.readData(String.class);
            }
        }
    };
    eventSource.register(listener);
    eventSource.open();
    serverSentEventService.send("/sseauth", data);

    //then
    await().atMost(2,  TimeUnit.SECONDS).untilAsserted(() -> assertThat(eventData, equalTo(data)));
    eventSource.close();
}
 

import org.glassfish.jersey.media.sse.InboundEvent; //导入依赖的package包/类
@Test
public void testSendDataWithInvalidAuthentication() throws InterruptedException {
    //given
    final ServerSentEventService serverSentEventService = Application.getInstance(ServerSentEventService.class);
    final Config config = Application.getInstance(Config.class);
    final String data = "Server sent data with authentication FTW!";

    //when
    final WebTarget target = ClientBuilder.newBuilder()
            .register(SseFeature.class)
            .build()
            .target("http://" + config.getConnectorHttpHost() + ":" + config.getConnectorHttpPort() + "/sseauth");

    final CustomWebTarget customWebTarget = new CustomWebTarget(target, new Cookie(COOKIE_NAME, INVALID_COOKIE_VALUE));
    final EventSource eventSource = EventSource.target(customWebTarget).build();
    final EventListener listener = new EventListener() {
        @Override
        public void onEvent(InboundEvent inboundEvent) {
            if (StringUtils.isBlank(eventData)) {
                eventData = inboundEvent.readData(String.class);
            }
        }
    };
    eventSource.register(listener);
    eventSource.open();
    serverSentEventService.send("/sseauth", data);

    //then
    await().atMost(2,  TimeUnit.SECONDS).untilAsserted(() -> assertThat(eventData, not(equalTo(data))));
    eventSource.close();
}
 

import org.glassfish.jersey.media.sse.InboundEvent; //导入依赖的package包/类
private void onEvent(InboundEvent inboundEvent) {
    try {
        lastEventTimestamp = System.currentTimeMillis();

        String name = inboundEvent.getName();
        String data = inboundEvent.readData();

        logger.debug("Received '{}' event, data: {}", name, data);

        if (!connected) {
            logger.debug("Connected to streaming events");
            connected = true;
            listeners.forEach(listener -> listener.onConnected());
        }

        if (AUTH_REVOKED.equals(name)) {
            logger.debug("API authorization has been revoked for access token: {}", data);
            listeners.forEach(listener -> listener.onAuthorizationRevoked(data));
        } else if (ERROR.equals(name)) {
            logger.warn("Error occurred: {}", data);
            listeners.forEach(listener -> listener.onError(data));
        } else if (KEEP_ALIVE.equals(name)) {
            logger.debug("Received message to keep connection alive");
        } else if (OPEN.equals(name)) {
            logger.debug("Event stream opened");
        } else if (PUT.equals(name)) {
            logger.debug("Data has changed (or initial data sent)");
            lastReceivedTopLevelData = gson.fromJson(data, TopLevelStreamingData.class).getData();
            listeners.forEach(listener -> listener.onNewTopLevelData(lastReceivedTopLevelData));
        } else {
            logger.debug("Received unhandled event with name '{}' and data '{}'", name, data);
        }
    } catch (Exception e) {
        // catch exceptions here otherwise they will be swallowed by the implementation
        logger.warn("An exception occurred while processing the inbound event", e);
    }
}
 

import org.glassfish.jersey.media.sse.InboundEvent; //导入依赖的package包/类
@Override
protected String doInBackground(Integer... params) {


    Client client = ClientBuilder.newBuilder()
            .register(SseFeature.class).build();

    String url = UtilityGeneral.getServiceURL(MyApplication.getAppContext()) + "/api/broadcast";

    WebTarget target = client.target(url);



    EventInput eventInput = target.request().get(EventInput.class);
    while (!eventInput.isClosed()) {
        final InboundEvent inboundEvent = eventInput.read();
        if (inboundEvent == null) {
            // connection has been closed
            break;
        }
        System.out.println(inboundEvent.getName() + "; "
                + inboundEvent.readData(String.class));

        publishProgress(inboundEvent.getName() + "; "
                + inboundEvent.readData(String.class));

    }

    return "finished";
}
 

import org.glassfish.jersey.media.sse.InboundEvent; //导入依赖的package包/类
void eventSourceThree()
    {


        eventSource = new EventSource(target) {
            @Override
            public void onEvent(InboundEvent inboundEvent) {

//                if ("message-to-client".equals(inboundEvent.getName())) {
                System.out.println(inboundEvent.getName() + "; "
                        + inboundEvent.readData(String.class));

                String eventName = inboundEvent.getName();
                String message = inboundEvent.readData(String.class);


                NotificationCompat.Builder mBuilder =
                        new NotificationCompat.Builder(IntentServiceSSE.this)
                                .setContentTitle(eventName)
                                .setContentText(message)
                                .setContentInfo(message)
                                .setSmallIcon(R.mipmap.shopping_basket_png)
                                .setSound(RingtoneManager.getDefaultUri(RingtoneManager.TYPE_NOTIFICATION));


                NotificationManager mNotificationManager =
                        (NotificationManager) getSystemService(Context.NOTIFICATION_SERVICE);

                mNotificationManager.notify(2, mBuilder.build());
            }


        };
    }
 

import org.glassfish.jersey.media.sse.InboundEvent; //导入依赖的package包/类
void handleNotificationThree()
    {
        Client client = ClientBuilder.newBuilder()
                .register(SseFeature.class).build();


        int endUserID = -1;

        if(UtilityLogin.getEndUser(getBaseContext())!=null)
        {
            endUserID = UtilityLogin.getEndUser(getBaseContext()).getEndUserID();
        }
        else
        {
            return;
        }


        System.out.println("On Handle Intent : Handle Notification !");

        String url = UtilityGeneral.getServiceURL(MyApplication.getAppContext()) + "/api/v1/EndUser/Notifications/" + String.valueOf(endUserID);

        System.out.println("URL : " + url);
//        logMessage("URL : " + url);


        WebTarget target = client.target(url);

        eventSourceThree = EventSource.target(target)
                .reconnectingEvery(5,TimeUnit.SECONDS)
                .build();



        EventListener listener = new EventListener() {
            @Override
            public void onEvent(InboundEvent inboundEvent) {


                    System.out.println(inboundEvent.getName() + "; "
                            + inboundEvent.readData(String.class));

                    String eventName = inboundEvent.getName();
                    String message = inboundEvent.readData(String.class);


                    NotificationCompat.Builder mBuilder =
                            new NotificationCompat.Builder(IntentServiceSSE.this)
                                    .setContentTitle(eventName)
                                    .setContentText(message)
                                    .setContentInfo(message)
                                    .setSmallIcon(R.mipmap.shopping_basket_png)
                                    .setSound(RingtoneManager.getDefaultUri(RingtoneManager.TYPE_NOTIFICATION));


                    NotificationManager mNotificationManager =
                            (NotificationManager) getSystemService(Context.NOTIFICATION_SERVICE);

                    mNotificationManager.notify(2, mBuilder.build());
            }
        };



        eventSourceThree.register(listener);
        eventSourceThree.open();


    }