RyaStreamsCommands implements CommandMarker { @CliCommand(value = STREAMS_CONFIGURE_CMD, help = "Connect a Rya Streams subsystem to a Rya Instance.") public String configureRyaStreams( @CliOption(key = {"kafkaHostname"}, mandatory = true, help = "The hostname of the Kafka Broker.") final String kafkaHostname, @CliOption(key = {"kafkaPort"}, mandatory = true, help = "The port of the Kafka Broker.") final int kafkaPort) { final Optional<RyaStreamsClient> oldClient = state.getShellState().getRyaStreamsCommands(); if(oldClient.isPresent()) { try { oldClient.get().close(); } catch (final Exception e) { System.err.print("Warning: Could not close the old Rya Streams Client."); e.printStackTrace(); } } final String ryaInstance = state.getShellState().getRyaInstanceName().get(); final RyaClient ryaClient = state.getShellState().getConnectedCommands().get(); try { final RyaStreamsDetails streamsDetails = new RyaStreamsDetails(kafkaHostname, kafkaPort); ryaClient.getSetRyaStreamsConfiguration().setRyaStreamsConfiguration(ryaInstance, streamsDetails); } catch (final RyaClientException e) { throw new RuntimeException("Could not update the Rya instance's Rya Details to include the new " + "information. This command failed to complete.", e); } final RyaStreamsClient newClient = KafkaRyaStreamsClientFactory.make(ryaInstance, kafkaHostname, kafkaPort); state.connectedToRyaStreams(newClient); if(oldClient.isPresent()) { return "The Rya Streams subsystem that this Rya instance uses has been changed. Any queries that were " + "maintained by the previous subsystem will need to be migrated to the new one."; } else { return "The Rya Instance has been updated to use the provided Rya Streams subsystem. " + "Rya Streams commands are now avaiable while connected to this instance."; } } @Autowired  RyaStreamsCommands(
            final SharedShellState state,
            final SparqlPrompt sparqlPrompt,
            final ConsolePrinter consolePrinter); @CliAvailabilityIndicator({ STREAMS_CONFIGURE_CMD, STREAMS_DETAILS_CMD}) boolean areConfigCommandsAvailable(); @CliAvailabilityIndicator({ STREAM_QUERIES_ADD_CMD, STREAM_QUERIES_DELETE_CMD, STREAM_QUERIES_START_CMD, STREAM_QUERIES_STOP_CMD, STREAM_QUERIES_LIST_CMD, STREAM_QUERIES_DETAILS_CMD}) boolean areQueriesCommandsAvailable(); @CliCommand(value = STREAMS_CONFIGURE_CMD, help = "Connect a Rya Streams subsystem to a Rya Instance.") String configureRyaStreams(
            @CliOption(key = {"kafkaHostname"}, mandatory = true, help = "The hostname of the Kafka Broker.")
            final String kafkaHostname,
            @CliOption(key = {"kafkaPort"}, mandatory = true, help = "The port of the Kafka Broker.")
            final int kafkaPort); @CliCommand(value = STREAMS_DETAILS_CMD, help = "Print information about which Rya Streams subsystem the Rya instance is connected to.") String printRyaStreamsDetails(); @CliCommand(value = STREAM_QUERIES_ADD_CMD, help = "Add a SPARQL query to the Rya Streams subsystem.") String addQuery(
            @CliOption(key = {"inactive"}, mandatory = false, unspecifiedDefaultValue = "false", specifiedDefaultValue = "true",
                       help = "Setting this flag will add the query, but not run it. (default: false)")
            final boolean inactive,
            @CliOption(key = {"insert"}, mandatory = false, unspecifiedDefaultValue = "false", specifiedDefaultValue = "true",
                       help = "Setting this flag will insert the query's results back into Rya. (default: false)")
            final boolean isInsert); @CliCommand(value = STREAM_QUERIES_DELETE_CMD, help = "Delete a SPARQL query from the Rya Streams subsystem.") String deleteQuery(
            @CliOption(key= {"queryId"}, mandatory = true, help = "The ID of the query to remove.")
            final String queryId); @CliCommand(value = STREAM_QUERIES_START_CMD, help = "Start processing a SPARQL query using the Rya Streams subsystem.") String startQuery(
            @CliOption(key= {"queryId"}, mandatory = true, help = "The ID of the query to start processing.")
            final String queryId); @CliCommand(value = STREAM_QUERIES_STOP_CMD, help = "Stop processing a SPARQL query using the Rya Streams subsystem.") String stopQuery(
            @CliOption(key= {"queryId"}, mandatory = true, help = "The ID of the query to stop processing.")
            final String queryId); @CliCommand(value = STREAM_QUERIES_LIST_CMD, help = "List the queries that are being managed by the configured Rya Streams subsystem.") String listQueries(); @CliCommand(value = STREAM_QUERIES_DETAILS_CMD, help = "Print detailed information about a specific query managed by the Rya Streams subsystem.") String printQueryDetails(
            @CliOption(key= {"queryId"}, mandatory = true, help = "The ID of the query whose details will be printed.")
            final String queryId); static final String STREAMS_CONFIGURE_CMD; static final String STREAMS_DETAILS_CMD; static final String STREAM_QUERIES_ADD_CMD; static final String STREAM_QUERIES_DELETE_CMD; static final String STREAM_QUERIES_START_CMD; static final String STREAM_QUERIES_STOP_CMD; static final String STREAM_QUERIES_LIST_CMD; static final String STREAM_QUERIES_DETAILS_CMD; }
@Test public void configureRyaStreams() throws Exception { final RyaClient mockCommands = mock(RyaClient.class); final SetRyaStreamsConfiguration setStreams = mock(SetRyaStreamsConfiguration.class); when(mockCommands.getSetRyaStreamsConfiguration()).thenReturn(setStreams); final SharedShellState state = new SharedShellState(); state.connectedToAccumulo(mock(AccumuloConnectionDetails.class), mockCommands); state.connectedToInstance("unitTest"); assertFalse(state.getShellState().getRyaStreamsCommands().isPresent()); try { final RyaStreamsCommands commands = new RyaStreamsCommands(state, mock(SparqlPrompt.class), mock(ConsolePrinter.class)); final String message = commands.configureRyaStreams("localhost", 6); final RyaStreamsDetails expectedDetails = new RyaStreamsDetails("localhost", 6); verify(setStreams).setRyaStreamsConfiguration(eq("unitTest"), eq(expectedDetails)); assertTrue(state.getShellState().getRyaStreamsCommands().isPresent()); final String expected = "The Rya Instance has been updated to use the provided Rya Streams subsystem. " + "Rya Streams commands are now avaiable while connected to this instance."; assertEquals(expected, message); } finally { state.getShellState().getRyaStreamsCommands().get().close(); } }