import org.glassfish.jersey.media.sse.SseFeature; //导入依赖的package包/类
protected void startServer() {
        WifiManager wifiMgr = (WifiManager) getApplicationContext()
                .getSystemService(Service.WIFI_SERVICE);
        if (wifiMgr.isWifiEnabled()) {
            // Deprecated. Does not support ipv6. *shrug* :)
            String ipAddress = Formatter.formatIpAddress(wifiMgr.getConnectionInfo()
                    .getIpAddress());

            URI baseUri = UriBuilder.fromUri("http://" + ipAddress)
                    .port(49152)
                    .build();
            ResourceConfig config = new ResourceConfig(SseFeature.class)
                    .register(JacksonFeature.class);
            config.registerInstances(new SecureFilter(this));
            config.registerInstances(new DeskDroidResource(this));
//            server = JettyHttpContainerFactory.createServer(baseUri, config);
            server = GrizzlyHttpServerFactory.createHttpServer(baseUri, config);
        }
    }
 

import org.glassfish.jersey.media.sse.SseFeature; //导入依赖的package包/类
@Override
public void run(InventoryItemApiConfiguration configuration, Environment environment) throws Exception {
    configureObjectMapper(environment);

    CommandDispatcher commandDispatcher = configuration.getCommandDispatcherFactory().build(environment);

    environment.jersey().register(new ApiListingResource());
    environment.jersey().register(SseFeature.class);

    InventoryItemResource resource = new InventoryItemResource(new InventoryItemsQuery(), commandDispatcher);
    environment.jersey().register(resource);
    environment.lifecycle().manage(new KafkaDenormalizer());
    environment.lifecycle().manage(new HazelcastManaged());

    StreamBroadcaster broadcaster = configuration.getStreamBroadcasterFactory().build(environment);
    broadcaster.addObserver(resource);

    configureSwagger(environment);
}
 

import org.glassfish.jersey.media.sse.SseFeature; //导入依赖的package包/类
@GET
@Path("/Notifications/{ShopID}")
@Produces(SseFeature.SERVER_SENT_EVENTS)
@RolesAllowed({GlobalConstants.ROLE_SHOP_ADMIN})
public EventOutput listenToBroadcast(@PathParam("ShopID")int shopID) {
	final EventOutput eventOutput = new EventOutput();

	if(Globals.broadcasterMap.get(shopID)!=null)
	{
		SseBroadcaster broadcasterOne = Globals.broadcasterMap.get(shopID);
		broadcasterOne.add(eventOutput);
	}
	else
	{
		SseBroadcaster broadcasterTwo = new SseBroadcaster();
		broadcasterTwo.add(eventOutput);
		Globals.broadcasterMap.put(shopID,broadcasterTwo);
	}

	return eventOutput;
}
 

import org.glassfish.jersey.media.sse.SseFeature; //导入依赖的package包/类
@GET
@Path("/Notifications/{ShopID}")
@Produces(SseFeature.SERVER_SENT_EVENTS)
public EventOutput listenToBroadcast(@PathParam("ShopID")int shopID) {
	final EventOutput eventOutput = new EventOutput();

	if(Globals.broadcasterMap.get(shopID)!=null)
	{
		SseBroadcaster broadcasterOne = Globals.broadcasterMap.get(shopID);
		broadcasterOne.add(eventOutput);
	}
	else
	{
		SseBroadcaster broadcasterTwo = new SseBroadcaster();
		broadcasterTwo.add(eventOutput);
		Globals.broadcasterMap.put(shopID,broadcasterTwo);
	}

	return eventOutput;
}
 

import org.glassfish.jersey.media.sse.SseFeature; //导入依赖的package包/类
@GET
@Path("/Notifications/{EndUserID}")
@Produces(SseFeature.SERVER_SENT_EVENTS)
public EventOutput listenToBroadcast(@PathParam("EndUserID")int endUserID) {

	final EventOutput eventOutput = new EventOutput();

	if(Globals.broadcasterMapEndUser.get(endUserID)!=null)
	{
		SseBroadcaster broadcasterOne = Globals.broadcasterMapEndUser.get(endUserID);
		broadcasterOne.add(eventOutput);
	}
	else
	{
		SseBroadcaster broadcasterTwo = new SseBroadcaster();
		broadcasterTwo.add(eventOutput);
		Globals.broadcasterMapEndUser.put(endUserID,broadcasterTwo);
	}

	return eventOutput;
}
 

import org.glassfish.jersey.media.sse.SseFeature; //导入依赖的package包/类
@GET
@Path("/{ShopID}")
@Produces(SseFeature.SERVER_SENT_EVENTS)
public EventOutput listenToBroadcast(@PathParam("ShopID")int shopID) {
    final EventOutput eventOutput = new EventOutput();

    if(Globals.broadcasterMap.get(shopID)!=null)
    {
        SseBroadcaster broadcasterOne = Globals.broadcasterMap.get(shopID);
        broadcasterOne.add(eventOutput);
    }
    else
    {
        SseBroadcaster broadcasterTwo = new SseBroadcaster();
        broadcasterTwo.add(eventOutput);
        Globals.broadcasterMap.put(shopID,broadcasterTwo);
    }
    return eventOutput;
}
 

import org.glassfish.jersey.media.sse.SseFeature; //导入依赖的package包/类
public <RootResponse> Response<RootResponse> create(@NonNull String url, @NonNull Class<RootResponse> clazz,
		ClientConfiguration clientConfigurationOrNull) {
	Optional<ClientConfiguration> clientConfiguration = Optional.ofNullable(clientConfigurationOrNull);
	JerseyClient newClient = jerseyClientBuilder.register(SseFeature.class).withConfig(clientConfig).build();
	newClient.register(LastEventIdHeaderFilter.class);

	if (clientConfigurationOrNull != null && !Strings.isNullOrEmpty(clientConfigurationOrNull.getAuthorization())) {
		newClient.register(new AuthHeaderFilter(clientConfigurationOrNull.getAuthorization()));
	}

	MediaType mediaType = clientConfiguration.map(c -> c.getMediaType()).orElse(MediaType.APPLICATION_JSON_TYPE);

	ResponseBuilder responseBuilder = new ResponseBuilder(newClient, objectMapper, mediaType);
	JerseyWebTarget webTarget = newClient.target(url);
	Builder requestBuilder = webTarget.request(mediaType);
	String readEntity = requestBuilder.get().readEntity(String.class);
	return responseBuilder.buildResponse(readEntity, clazz, URI.create(url)).get();
}
 

import org.glassfish.jersey.media.sse.SseFeature; //导入依赖的package包/类
@Test
public void testCreate() throws Exception {
	when(jerseyClientBuilder.build()).thenReturn(client);
	when(jerseyClientBuilder.register(SseFeature.class)).thenReturn(jerseyClientBuilder);
	when(jerseyClientBuilder.withConfig(any())).thenReturn(jerseyClientBuilder);
	when(client.target(anyString())).thenReturn(webTarget);
	when(webTarget.request(any(MediaType.class))).thenReturn(builder);
	when(builder.get()).thenReturn(response);
	when(response.readEntity(String.class)).thenReturn("");

	uut.create("http://mercateo.com/test", Object.class);
	verify(jerseyClientBuilder).build();
	verify(client).target(anyString());
	verify(webTarget).request(any(MediaType.class));
	verify(builder).get();
	verify(response).readEntity(String.class);
}
 

import org.glassfish.jersey.media.sse.SseFeature; //导入依赖的package包/类
@Test
public void testCreateWithConfig() throws Exception {
	when(jerseyClientBuilder.build()).thenReturn(client);
	when(jerseyClientBuilder.register(SseFeature.class)).thenReturn(jerseyClientBuilder);
	when(jerseyClientBuilder.withConfig(any())).thenReturn(jerseyClientBuilder);
	when(client.target(anyString())).thenReturn(webTarget);
	when(webTarget.request(any(MediaType.class))).thenReturn(builder);
	when(builder.get()).thenReturn(response);
	when(response.readEntity(String.class)).thenReturn("");

	ClientConfiguration clientConfiguration = new ClientConfiguration("test", null);

	uut.create("http://mercateo.com/test", Object.class, clientConfiguration);
	verify(jerseyClientBuilder).build();
	verify(client).target(anyString());
	verify(webTarget).request(any(MediaType.class));
	verify(builder).get();
	verify(response).readEntity(String.class);
	verify(client).register(eq(new AuthHeaderFilter("test")));
}
 

import org.glassfish.jersey.media.sse.SseFeature; //导入依赖的package包/类
@GET
@Path("events")
@Produces(SseFeature.SERVER_SENT_EVENTS)
public EventOutput events() {
    final EventOutput eventOutput = new EventOutput();

    if (!broadcaster.add(eventOutput)) {
        // 503 -> 5s delayed client reconnect attempt.
        throw new ServiceUnavailableException(5L);
    }

    try {
        eventOutput.write(event());
    } catch (final IOException ioe) {
        // NO-OP.
    }

    return eventOutput;
}
 

import org.glassfish.jersey.media.sse.SseFeature; //导入依赖的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.SseFeature; //导入依赖的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.SseFeature; //导入依赖的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.SseFeature; //导入依赖的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.SseFeature; //导入依赖的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.SseFeature; //导入依赖的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.SseFeature; //导入依赖的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.SseFeature; //导入依赖的package包/类
@Inject
public WebApp(Injector injector, ServiceLocator serviceLocator) {
    log.info("Setting up Web Application");

    register(SseFeature.class);
    register(MultiPartFeature.class);
    register(ExposrExceptionMapper.class);

    // Resources.
    register(BuildResource.class);
    register(ExposrResource.class);
    register(ProjectResource.class);
    register(SyncResource.class);
    register(TaskResource.class);

    GuiceBridge.getGuiceBridge().initializeGuiceBridge(serviceLocator);

    final GuiceIntoHK2Bridge bridge = serviceLocator
            .getService(GuiceIntoHK2Bridge.class);

    bridge.bridgeGuiceInjector(injector);
}
 

import org.glassfish.jersey.media.sse.SseFeature; //导入依赖的package包/类
@GET
@PermitAll
@Produces(SseFeature.SERVER_SENT_EVENTS)
public EventOutput listenToEvents(@QueryParam("token") String token) {
  String removedToken = SLTRegister.remove(token);

  if (removedToken == null) {
    return null;
  }

  final EventOutput eventOutput = new EventOutput();
  sseBroadcaster.add(eventOutput);
  return eventOutput;
}
 

import org.glassfish.jersey.media.sse.SseFeature; //导入依赖的package包/类
@GET
@Produces(SseFeature.SERVER_SENT_EVENTS)
@NoCache
@Path("id-only")
public EventOutput getServerSentEvents(
        @NotNull @Valid @BeanParam SubscriptionRequestParams subscriptionRequestParams) {
    return createEventOutput(subscriptionRequestParams, false);
}