test_class.declarator: operations
test_class.modifier: @Mock
test_class.original_string: @Mock
  	KafkaOps operations;
test_class.type: KafkaOps
test_class.var_name: operations
test_class.declarator: producerChannels
test_class.modifier: @Mock
test_class.original_string: @Mock
  	CSourceProducerChannel producerChannels;
test_class.type: CSourceProducerChannel
test_class.var_name: producerChannels
test_class.declarator: csourceService
test_class.modifier: @InjectMocks
test_class.original_string: @InjectMocks
  	CSourceService csourceService;
test_class.type: CSourceService
test_class.var_name: csourceService
test_class.declarator: objectMapper
test_class.modifier: @MockBean
test_class.original_string: @MockBean
  	ObjectMapper objectMapper;
test_class.type: ObjectMapper
test_class.var_name: objectMapper
test_class.declarator: csourceSubService
test_class.modifier: @Mock
test_class.original_string: @Mock
  	CSourceSubscriptionService csourceSubService;
test_class.type: CSourceSubscriptionService
test_class.var_name: csourceSubService
test_class.declarator: csourceReg
test_class.modifier: private
test_class.original_string: private CSourceRegistration csourceReg;
test_class.type: CSourceRegistration
test_class.var_name: csourceReg
test_class.declarator: updateCSourceReg
test_class.modifier: private
test_class.original_string: private CSourceRegistration updateCSourceReg;
test_class.type: CSourceRegistration
test_class.var_name: updateCSourceReg
test_class.declarator: payload
test_class.original_string: String payload;
test_class.type: String
test_class.var_name: payload
test_class.declarator: updatePayload
test_class.original_string: String updatePayload;
test_class.type: String
test_class.var_name: updatePayload
test_class.declarator: blankNode
test_class.original_string: JsonNode blankNode;
test_class.type: JsonNode
test_class.var_name: blankNode
test_class.declarator: payloadNode
test_class.original_string: JsonNode payloadNode;
test_class.type: JsonNode
test_class.var_name: payloadNode
test_class.file: Registry/RegistryManager/src/test/java/eu/neclab/ngsildbroker/registryhandler/service/CSourceServiceTest.java
test_class.identifier: CSourceServiceTest
test_case.body: @Test
  	public void registerCSourceTest(){
  		try {
  			csourceReg.setInternal(true);
  			Mockito.doReturn(false).when(operations).isMessageExists(any(), any());
  			URI uri = csourceService.registerCSource(csourceReg);
  			Assert.assertEquals(uri, new URI("urn:ngsi-ld:ContextSourceRegistration:csr1a3456"));
  		}catch(Exception ex) {
  			Assert.fail();
  		}
  	}
test_case.class_method_signature: CSourceServiceTest.registerCSourceTest()
test_case.full_signature: @Test public void registerCSourceTest()
test_case.identifier: registerCSourceTest
test_case.invocations: setInternal isMessageExists when doReturn any any registerCSource assertEquals fail
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void registerCSourceTest()
focal_class.declarator: logger = LoggerFactory.getLogger(RegistryController.class)
focal_class.modifier: private final static
focal_class.original_string: private final static Logger logger = LoggerFactory.getLogger(RegistryController.class);
focal_class.type: Logger
focal_class.var_name: logger
focal_class.declarator: BOOTSTRAP_SERVERS
focal_class.modifier: @Value("${bootstrap.servers}")
focal_class.original_string: @Value("${bootstrap.servers}")
  	String BOOTSTRAP_SERVERS;
focal_class.type: String
focal_class.var_name: BOOTSTRAP_SERVERS
focal_class.declarator: operations
focal_class.modifier: @Autowired
  	@Qualifier("rmops")
focal_class.original_string: @Autowired
  	@Qualifier("rmops")
  	KafkaOps operations;
focal_class.type: KafkaOps
focal_class.var_name: operations
focal_class.declarator: objectMapper
focal_class.modifier: @Autowired
focal_class.original_string: @Autowired
  	ObjectMapper objectMapper;
focal_class.type: ObjectMapper
focal_class.var_name: objectMapper
focal_class.declarator: startupConfig
focal_class.modifier: @Autowired
focal_class.original_string: @Autowired
  	StartupConfig startupConfig;
focal_class.type: StartupConfig
focal_class.var_name: startupConfig
focal_class.declarator: csourceDAO
focal_class.modifier: @Autowired
  	@Qualifier("rmcsourcedao")
focal_class.original_string: @Autowired
  	@Qualifier("rmcsourcedao")
  	CSourceDAO csourceDAO;
focal_class.type: CSourceDAO
focal_class.var_name: csourceDAO
focal_class.declarator: csourceSubService
focal_class.modifier: @Autowired
focal_class.original_string: @Autowired
  	CSourceSubscriptionService csourceSubService;
focal_class.type: CSourceSubscriptionService
focal_class.var_name: csourceSubService
focal_class.declarator: queryParser
focal_class.modifier: @Autowired
  	@Qualifier("rmqueryParser")
focal_class.original_string: @Autowired
  	@Qualifier("rmqueryParser")
  	QueryParser queryParser;
focal_class.type: QueryParser
focal_class.var_name: queryParser
focal_class.declarator: CSOURCE_TOPIC
focal_class.modifier: @Value("${csource.source.topic}")
focal_class.original_string: @Value("${csource.source.topic}")
  	String CSOURCE_TOPIC;
focal_class.type: String
focal_class.var_name: CSOURCE_TOPIC
focal_class.declarator: producerChannels
focal_class.modifier: private final
focal_class.original_string: private final CSourceProducerChannel producerChannels;
focal_class.type: CSourceProducerChannel
focal_class.var_name: producerChannels
focal_class.declarator: regId2TimerTask = new HashMap<String, TimerTask>()
focal_class.original_string: HashMap<String, TimerTask> regId2TimerTask = new HashMap<String, TimerTask>();
focal_class.type: HashMap<String, TimerTask>
focal_class.var_name: regId2TimerTask
focal_class.declarator: watchDog = new Timer(true)
focal_class.original_string: Timer watchDog = new Timer(true);
focal_class.type: Timer
focal_class.var_name: watchDog
focal_class.declarator: workQueue = new ArrayBlockingQueue<Runnable>(50000, true)
focal_class.modifier: private
focal_class.original_string: private ArrayBlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<Runnable>(50000, true);
focal_class.type: ArrayBlockingQueue<Runnable>
focal_class.var_name: workQueue
focal_class.declarator: executor = new ThreadPoolExecutor(20, 50, 600000, TimeUnit.MILLISECONDS, workQueue)
focal_class.original_string: ThreadPoolExecutor executor = new ThreadPoolExecutor(20, 50, 600000, TimeUnit.MILLISECONDS, workQueue);
focal_class.type: ThreadPoolExecutor
focal_class.var_name: executor
focal_class.file: Registry/RegistryManager/src/main/java/eu/neclab/ngsildbroker/registryhandler/service/CSourceService.java
focal_class.identifier: CSourceService
focal_class.class_method_signature: CSourceService.CSourceService(CSourceProducerChannel producerChannels)
focal_class.full_signature:   CSourceService(CSourceProducerChannel producerChannels)
focal_class.identifier: CSourceService
focal_class.parameters: (CSourceProducerChannel producerChannels)
focal_class.signature:  CSourceService(CSourceProducerChannel producerChannels)
focal_class.class_method_signature: CSourceService.getCSourceRegistrations()
focal_class.full_signature: public List<JsonNode> getCSourceRegistrations()
focal_class.identifier: getCSourceRegistrations
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: List<JsonNode>
focal_class.signature: List<JsonNode> getCSourceRegistrations()
focal_class.class_method_signature: CSourceService.getCSourceRegistrationById(String registrationId)
focal_class.full_signature: public CSourceRegistration getCSourceRegistrationById(String registrationId)
focal_class.identifier: getCSourceRegistrationById
focal_class.modifiers: public
focal_class.parameters: (String registrationId)
focal_class.return: CSourceRegistration
focal_class.signature: CSourceRegistration getCSourceRegistrationById(String registrationId)
focal_class.class_method_signature: CSourceService.updateCSourceRegistration(String registrationId, String payload)
focal_class.full_signature: public boolean updateCSourceRegistration(String registrationId, String payload)
focal_class.identifier: updateCSourceRegistration
focal_class.modifiers: public
focal_class.parameters: (String registrationId, String payload)
focal_class.return: boolean
focal_class.signature: boolean updateCSourceRegistration(String registrationId, String payload)
focal_class.class_method_signature: CSourceService.handleFed()
focal_class.full_signature: private void handleFed()
focal_class.identifier: handleFed
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void handleFed()
focal_class.class_method_signature: CSourceService.registerCSource(CSourceRegistration csourceRegistration)
focal_class.full_signature: public URI registerCSource(CSourceRegistration csourceRegistration)
focal_class.identifier: registerCSource
focal_class.modifiers: public
focal_class.parameters: (CSourceRegistration csourceRegistration)
focal_class.return: URI
focal_class.signature: URI registerCSource(CSourceRegistration csourceRegistration)
focal_class.class_method_signature: CSourceService.generateUniqueRegId(CSourceRegistration csourceRegistration)
focal_class.full_signature: private URI generateUniqueRegId(CSourceRegistration csourceRegistration)
focal_class.identifier: generateUniqueRegId
focal_class.modifiers: private
focal_class.parameters: (CSourceRegistration csourceRegistration)
focal_class.return: URI
focal_class.signature: URI generateUniqueRegId(CSourceRegistration csourceRegistration)
focal_class.class_method_signature: CSourceService.csourceTimerTask(CSourceRegistration csourceReg)
focal_class.full_signature: public void csourceTimerTask(CSourceRegistration csourceReg)
focal_class.identifier: csourceTimerTask
focal_class.modifiers: public
focal_class.parameters: (CSourceRegistration csourceReg)
focal_class.return: void
focal_class.signature: void csourceTimerTask(CSourceRegistration csourceReg)
focal_class.class_method_signature: CSourceService.deleteCSourceRegistration(String registrationId)
focal_class.full_signature: public boolean deleteCSourceRegistration(String registrationId)
focal_class.identifier: deleteCSourceRegistration
focal_class.modifiers: public
focal_class.parameters: (String registrationId)
focal_class.return: boolean
focal_class.signature: boolean deleteCSourceRegistration(String registrationId)
focal_class.class_method_signature: CSourceService.handleEntityRegistration(Message<?> message)
focal_class.full_signature: @KafkaListener(topics = "${csource.registry.topic}", groupId = "regmanger") // (CSourceConsumerChannel.contextRegistryReadChannel) public void handleEntityRegistration(Message<?> message)
focal_class.identifier: handleEntityRegistration
focal_class.modifiers: @KafkaListener(topics = "${csource.registry.topic}", groupId = "regmanger") // (CSourceConsumerChannel.contextRegistryReadChannel) public
focal_class.parameters: (Message<?> message)
focal_class.return: void
focal_class.signature: void handleEntityRegistration(Message<?> message)
focal_class.class_method_signature: CSourceService.isValidURL(String urlString)
focal_class.full_signature: private static boolean isValidURL(String urlString)
focal_class.identifier: isValidURL
focal_class.modifiers: private static
focal_class.parameters: (String urlString)
focal_class.return: boolean
focal_class.signature: boolean isValidURL(String urlString)
focal_class.class_method_signature: CSourceService.isValidFutureDate(Long date)
focal_class.full_signature: private boolean isValidFutureDate(Long date)
focal_class.identifier: isValidFutureDate
focal_class.modifiers: private
focal_class.parameters: (Long date)
focal_class.return: boolean
focal_class.signature: boolean isValidFutureDate(Long date)
focal_class.class_method_signature: CSourceService.handleContextQuery(@Payload byte[] message)
focal_class.full_signature: @KafkaListener(topics = "${csource.query.topic}", groupId = "csourceQueryHandler") @SendTo // @SendTo("QUERY_RESULT") // for tests without QueryManager public byte[] handleContextQuery(@Payload byte[] message)
focal_class.identifier: handleContextQuery
focal_class.modifiers: @KafkaListener(topics = "${csource.query.topic}", groupId = "csourceQueryHandler") @SendTo // @SendTo("QUERY_RESULT") // for tests without QueryManager public
focal_class.parameters: (@Payload byte[] message)
focal_class.return: byte[]
focal_class.signature: byte[] handleContextQuery(@Payload byte[] message)
focal_method.body: public URI registerCSource(CSourceRegistration csourceRegistration) throws ResponseException, Exception {
  		MessageChannel messageChannel = producerChannels.csourceWriteChannel();
  		String id;
  		URI idUri = csourceRegistration.getId();
  		if (idUri == null) {
  			idUri = generateUniqueRegId(csourceRegistration);
  			csourceRegistration.setId(idUri);
  
  		}
  		id = idUri.toString();
  
  		if (csourceRegistration.getType() == null) {
  			logger.error("Invalid type!");
  			throw new ResponseException(ErrorType.BadRequestData);
  		}
  		if (!isValidURL(csourceRegistration.getEndpoint().toString())) {
  			logger.error("Invalid endpoint URL!");
  			throw new ResponseException(ErrorType.BadRequestData);
  		}
  		if (csourceRegistration.getInformation() == null) {
  			logger.error("Information is empty!");
  			throw new ResponseException(ErrorType.BadRequestData);
  		}
  		if (csourceRegistration.getExpires() != null && !isValidFutureDate(csourceRegistration.getExpires())) {
  			logger.error("Invalid expire date!");
  			throw new ResponseException(ErrorType.BadRequestData);
  		}
  //TODO replace this with a database only attempt
  		if (this.operations.isMessageExists(id, this.CSOURCE_TOPIC)) {
  			byte[] messageBytes = this.operations.getMessage(id, this.CSOURCE_TOPIC);
  			JsonNode messgeJson = objectMapper.createObjectNode();
  			messgeJson = objectMapper.readTree(messageBytes);
  			if (!messgeJson.isNull()) {
  				throw new ResponseException(ErrorType.AlreadyExists);
  			}
  		}
  
  		// TODO: [check for valid identifier (id)]
  		operations.pushToKafka(messageChannel, id.getBytes(), DataSerializer.toJson(csourceRegistration).getBytes());
  
  		this.csourceTimerTask(csourceRegistration);
  		if (!csourceRegistration.isInternal()) {
  			csourceSubService.checkSubscriptions(csourceRegistration, TriggerReason.newlyMatching);
  		}
  		handleFed();
  		return idUri;
  	}
focal_method.class_method_signature: CSourceService.registerCSource(CSourceRegistration csourceRegistration)
focal_method.full_signature: public URI registerCSource(CSourceRegistration csourceRegistration)
focal_method.identifier: registerCSource
focal_method.invocations: csourceWriteChannel getId generateUniqueRegId setId toString getType error isValidURL toString getEndpoint error getInformation error getExpires isValidFutureDate getExpires error isMessageExists getMessage createObjectNode readTree isNull pushToKafka getBytes getBytes toJson csourceTimerTask isInternal checkSubscriptions handleFed
focal_method.modifiers: public
focal_method.parameters: (CSourceRegistration csourceRegistration)
focal_method.return: URI
focal_method.signature: URI registerCSource(CSourceRegistration csourceRegistration)