@Test @Ignore public void testSendData() throws MarshalException, ValidationException, IOException, FilterParseException, SAXException, SQLException, RTCException { InputStream stream = ConfigurationTestUtils.getInputStreamForResource(this, "/org/opennms/netmgt/config/rtc-configuration.xml"); RTCConfigFactory.setInstance(new RTCConfigFactory(stream)); stream.close(); Resource categoryResource = ConfigurationTestUtils.getSpringResourceForResource(this, "/org/opennms/netmgt/config/categories.xml"); CategoryFactory.setInstance(new CategoryFactory(categoryResource)); stream = ConfigurationTestUtils.getInputStreamForResource(this, "/org/opennms/netmgt/config/test-database-schema.xml"); DatabaseSchemaConfigFactory.setInstance(new DatabaseSchemaConfigFactory(stream)); stream.close(); DataManager dataManager = new DataManager(); RTCManager.setDataManager(dataManager); String categoryName = "Database Servers"; String categoryNameUrl = "Database+Servers"; Category category = new Category(); category.setLabel(categoryName); category.setComment("Some database servers. Exciting, eh?"); category.setNormal(99.0); category.setWarning(97.0); RTCCategory rtcCategory = new RTCCategory(category, categoryName); Map<String, RTCCategory> rtcCategories = new HashMap<String, RTCCategory>(); rtcCategories.put(categoryName, rtcCategory); DataSender sender = new DataSender(rtcCategories, 1); sender.subscribe("http: sender.sendData(); }
public synchronized void sendData() { log().debug("In DataSender sendData()"); for (RTCCategory cat : m_categories.values()) { String catlabel = cat.getLabel(); if (log().isDebugEnabled()) log().debug("DataSender:sendData(): Category \'" + catlabel); Set<HttpPostInfo> urlList = m_catUrlMap.get(catlabel); if (urlList == null || urlList.size() <= 0) { if (log().isDebugEnabled()) log().debug("DataSender: category \'" + catlabel + "\' has no listeners"); continue; } if (log().isDebugEnabled()) log().debug("DataSender: category \'" + catlabel + "\' has listeners - converting to xml..."); int oldPriority = setCurrentThreadPriority(Thread.MAX_PRIORITY); EuiLevel euidata = null; try { euidata = m_euiMapper.convertToEuiLevelXML(cat); } catch (java.lang.OutOfMemoryError oe) { log().warn("DataSender: unable to convert data to xml for category: " + catlabel, oe); setCurrentThreadPriority(Thread.NORM_PRIORITY); continue; } catch (Throwable t) { log().warn("DataSender: unable to convert data to xml for category: " + catlabel, t); setCurrentThreadPriority(Thread.NORM_PRIORITY); } if (urlList != null && urlList.size() > 0) { Iterator<HttpPostInfo> urlIter = urlList.iterator(); while (urlIter.hasNext()) { HttpPostInfo postInfo = urlIter.next(); Reader inr = null; InputStream inp = null; try { inr = new PipedMarshaller(euidata).getReader(); if (log().isDebugEnabled()) log().debug("DataSender: posting data to: " + postInfo.getURLString()); inp = HttpUtils.post(postInfo.getURL(), inr, postInfo.getUser(), postInfo.getPassword(), 8 * HttpUtils.DEFAULT_POST_BUFFER_SIZE); if (log().isDebugEnabled()) log().debug("DataSender: posted data for category: " + catlabel); byte[] tmp = new byte[1024]; int bytesRead; while ((bytesRead = inp.read(tmp)) != -1) { if (log().isDebugEnabled()) { if (bytesRead > 0) log().debug("DataSender: post response: " + new String(tmp, 0, bytesRead)); } } postInfo.clearErrors(); } catch (IOException e) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + e.getClass().getName() + ": " + e.getMessage(), e); postInfo.incrementErrors(); setCurrentThreadPriority(Thread.NORM_PRIORITY); } catch (java.lang.OutOfMemoryError e) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + e.getClass().getName() + ": " + e.getMessage(), e); setCurrentThreadPriority(Thread.NORM_PRIORITY); } catch (RuntimeException e) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + e.getClass().getName() + ": " + e.getMessage(), e); setCurrentThreadPriority(Thread.NORM_PRIORITY); } catch (Throwable t) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + t.getClass().getName() + ": " + t.getMessage(), t); setCurrentThreadPriority(Thread.NORM_PRIORITY); } finally { IOUtils.closeQuietly(inp); IOUtils.closeQuietly(inr); } if (POST_ERROR_LIMIT > 0 && postInfo.getErrors() >= POST_ERROR_LIMIT) { urlIter.remove(); log().warn("URL " + postInfo.getURLString() + " UNSUBSCRIBED due to reaching error limit " + postInfo.getErrors()); } } } oldPriority = setCurrentThreadPriority(oldPriority); } }
DataSender implements Fiber { public synchronized void sendData() { log().debug("In DataSender sendData()"); for (RTCCategory cat : m_categories.values()) { String catlabel = cat.getLabel(); if (log().isDebugEnabled()) log().debug("DataSender:sendData(): Category \'" + catlabel); Set<HttpPostInfo> urlList = m_catUrlMap.get(catlabel); if (urlList == null || urlList.size() <= 0) { if (log().isDebugEnabled()) log().debug("DataSender: category \'" + catlabel + "\' has no listeners"); continue; } if (log().isDebugEnabled()) log().debug("DataSender: category \'" + catlabel + "\' has listeners - converting to xml..."); int oldPriority = setCurrentThreadPriority(Thread.MAX_PRIORITY); EuiLevel euidata = null; try { euidata = m_euiMapper.convertToEuiLevelXML(cat); } catch (java.lang.OutOfMemoryError oe) { log().warn("DataSender: unable to convert data to xml for category: " + catlabel, oe); setCurrentThreadPriority(Thread.NORM_PRIORITY); continue; } catch (Throwable t) { log().warn("DataSender: unable to convert data to xml for category: " + catlabel, t); setCurrentThreadPriority(Thread.NORM_PRIORITY); } if (urlList != null && urlList.size() > 0) { Iterator<HttpPostInfo> urlIter = urlList.iterator(); while (urlIter.hasNext()) { HttpPostInfo postInfo = urlIter.next(); Reader inr = null; InputStream inp = null; try { inr = new PipedMarshaller(euidata).getReader(); if (log().isDebugEnabled()) log().debug("DataSender: posting data to: " + postInfo.getURLString()); inp = HttpUtils.post(postInfo.getURL(), inr, postInfo.getUser(), postInfo.getPassword(), 8 * HttpUtils.DEFAULT_POST_BUFFER_SIZE); if (log().isDebugEnabled()) log().debug("DataSender: posted data for category: " + catlabel); byte[] tmp = new byte[1024]; int bytesRead; while ((bytesRead = inp.read(tmp)) != -1) { if (log().isDebugEnabled()) { if (bytesRead > 0) log().debug("DataSender: post response: " + new String(tmp, 0, bytesRead)); } } postInfo.clearErrors(); } catch (IOException e) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + e.getClass().getName() + ": " + e.getMessage(), e); postInfo.incrementErrors(); setCurrentThreadPriority(Thread.NORM_PRIORITY); } catch (java.lang.OutOfMemoryError e) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + e.getClass().getName() + ": " + e.getMessage(), e); setCurrentThreadPriority(Thread.NORM_PRIORITY); } catch (RuntimeException e) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + e.getClass().getName() + ": " + e.getMessage(), e); setCurrentThreadPriority(Thread.NORM_PRIORITY); } catch (Throwable t) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + t.getClass().getName() + ": " + t.getMessage(), t); setCurrentThreadPriority(Thread.NORM_PRIORITY); } finally { IOUtils.closeQuietly(inp); IOUtils.closeQuietly(inr); } if (POST_ERROR_LIMIT > 0 && postInfo.getErrors() >= POST_ERROR_LIMIT) { urlIter.remove(); log().warn("URL " + postInfo.getURLString() + " UNSUBSCRIBED due to reaching error limit " + postInfo.getErrors()); } } } oldPriority = setCurrentThreadPriority(oldPriority); } } }
DataSender implements Fiber { public synchronized void sendData() { log().debug("In DataSender sendData()"); for (RTCCategory cat : m_categories.values()) { String catlabel = cat.getLabel(); if (log().isDebugEnabled()) log().debug("DataSender:sendData(): Category \'" + catlabel); Set<HttpPostInfo> urlList = m_catUrlMap.get(catlabel); if (urlList == null || urlList.size() <= 0) { if (log().isDebugEnabled()) log().debug("DataSender: category \'" + catlabel + "\' has no listeners"); continue; } if (log().isDebugEnabled()) log().debug("DataSender: category \'" + catlabel + "\' has listeners - converting to xml..."); int oldPriority = setCurrentThreadPriority(Thread.MAX_PRIORITY); EuiLevel euidata = null; try { euidata = m_euiMapper.convertToEuiLevelXML(cat); } catch (java.lang.OutOfMemoryError oe) { log().warn("DataSender: unable to convert data to xml for category: " + catlabel, oe); setCurrentThreadPriority(Thread.NORM_PRIORITY); continue; } catch (Throwable t) { log().warn("DataSender: unable to convert data to xml for category: " + catlabel, t); setCurrentThreadPriority(Thread.NORM_PRIORITY); } if (urlList != null && urlList.size() > 0) { Iterator<HttpPostInfo> urlIter = urlList.iterator(); while (urlIter.hasNext()) { HttpPostInfo postInfo = urlIter.next(); Reader inr = null; InputStream inp = null; try { inr = new PipedMarshaller(euidata).getReader(); if (log().isDebugEnabled()) log().debug("DataSender: posting data to: " + postInfo.getURLString()); inp = HttpUtils.post(postInfo.getURL(), inr, postInfo.getUser(), postInfo.getPassword(), 8 * HttpUtils.DEFAULT_POST_BUFFER_SIZE); if (log().isDebugEnabled()) log().debug("DataSender: posted data for category: " + catlabel); byte[] tmp = new byte[1024]; int bytesRead; while ((bytesRead = inp.read(tmp)) != -1) { if (log().isDebugEnabled()) { if (bytesRead > 0) log().debug("DataSender: post response: " + new String(tmp, 0, bytesRead)); } } postInfo.clearErrors(); } catch (IOException e) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + e.getClass().getName() + ": " + e.getMessage(), e); postInfo.incrementErrors(); setCurrentThreadPriority(Thread.NORM_PRIORITY); } catch (java.lang.OutOfMemoryError e) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + e.getClass().getName() + ": " + e.getMessage(), e); setCurrentThreadPriority(Thread.NORM_PRIORITY); } catch (RuntimeException e) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + e.getClass().getName() + ": " + e.getMessage(), e); setCurrentThreadPriority(Thread.NORM_PRIORITY); } catch (Throwable t) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + t.getClass().getName() + ": " + t.getMessage(), t); setCurrentThreadPriority(Thread.NORM_PRIORITY); } finally { IOUtils.closeQuietly(inp); IOUtils.closeQuietly(inr); } if (POST_ERROR_LIMIT > 0 && postInfo.getErrors() >= POST_ERROR_LIMIT) { urlIter.remove(); log().warn("URL " + postInfo.getURLString() + " UNSUBSCRIBED due to reaching error limit " + postInfo.getErrors()); } } } oldPriority = setCurrentThreadPriority(oldPriority); } } DataSender(Map<String, RTCCategory> categories, int numSenders); }
DataSender implements Fiber { public synchronized void sendData() { log().debug("In DataSender sendData()"); for (RTCCategory cat : m_categories.values()) { String catlabel = cat.getLabel(); if (log().isDebugEnabled()) log().debug("DataSender:sendData(): Category \'" + catlabel); Set<HttpPostInfo> urlList = m_catUrlMap.get(catlabel); if (urlList == null || urlList.size() <= 0) { if (log().isDebugEnabled()) log().debug("DataSender: category \'" + catlabel + "\' has no listeners"); continue; } if (log().isDebugEnabled()) log().debug("DataSender: category \'" + catlabel + "\' has listeners - converting to xml..."); int oldPriority = setCurrentThreadPriority(Thread.MAX_PRIORITY); EuiLevel euidata = null; try { euidata = m_euiMapper.convertToEuiLevelXML(cat); } catch (java.lang.OutOfMemoryError oe) { log().warn("DataSender: unable to convert data to xml for category: " + catlabel, oe); setCurrentThreadPriority(Thread.NORM_PRIORITY); continue; } catch (Throwable t) { log().warn("DataSender: unable to convert data to xml for category: " + catlabel, t); setCurrentThreadPriority(Thread.NORM_PRIORITY); } if (urlList != null && urlList.size() > 0) { Iterator<HttpPostInfo> urlIter = urlList.iterator(); while (urlIter.hasNext()) { HttpPostInfo postInfo = urlIter.next(); Reader inr = null; InputStream inp = null; try { inr = new PipedMarshaller(euidata).getReader(); if (log().isDebugEnabled()) log().debug("DataSender: posting data to: " + postInfo.getURLString()); inp = HttpUtils.post(postInfo.getURL(), inr, postInfo.getUser(), postInfo.getPassword(), 8 * HttpUtils.DEFAULT_POST_BUFFER_SIZE); if (log().isDebugEnabled()) log().debug("DataSender: posted data for category: " + catlabel); byte[] tmp = new byte[1024]; int bytesRead; while ((bytesRead = inp.read(tmp)) != -1) { if (log().isDebugEnabled()) { if (bytesRead > 0) log().debug("DataSender: post response: " + new String(tmp, 0, bytesRead)); } } postInfo.clearErrors(); } catch (IOException e) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + e.getClass().getName() + ": " + e.getMessage(), e); postInfo.incrementErrors(); setCurrentThreadPriority(Thread.NORM_PRIORITY); } catch (java.lang.OutOfMemoryError e) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + e.getClass().getName() + ": " + e.getMessage(), e); setCurrentThreadPriority(Thread.NORM_PRIORITY); } catch (RuntimeException e) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + e.getClass().getName() + ": " + e.getMessage(), e); setCurrentThreadPriority(Thread.NORM_PRIORITY); } catch (Throwable t) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + t.getClass().getName() + ": " + t.getMessage(), t); setCurrentThreadPriority(Thread.NORM_PRIORITY); } finally { IOUtils.closeQuietly(inp); IOUtils.closeQuietly(inr); } if (POST_ERROR_LIMIT > 0 && postInfo.getErrors() >= POST_ERROR_LIMIT) { urlIter.remove(); log().warn("URL " + postInfo.getURLString() + " UNSUBSCRIBED due to reaching error limit " + postInfo.getErrors()); } } } oldPriority = setCurrentThreadPriority(oldPriority); } } DataSender(Map<String, RTCCategory> categories, int numSenders); synchronized void start(); synchronized void stop(); String getName(); int getStatus(); synchronized void subscribe(String url, String catlabel, String user, String passwd); synchronized void unsubscribe(final String urlStr); synchronized void sendData(); synchronized void notifyToSend(); }
DataSender implements Fiber { public synchronized void sendData() { log().debug("In DataSender sendData()"); for (RTCCategory cat : m_categories.values()) { String catlabel = cat.getLabel(); if (log().isDebugEnabled()) log().debug("DataSender:sendData(): Category \'" + catlabel); Set<HttpPostInfo> urlList = m_catUrlMap.get(catlabel); if (urlList == null || urlList.size() <= 0) { if (log().isDebugEnabled()) log().debug("DataSender: category \'" + catlabel + "\' has no listeners"); continue; } if (log().isDebugEnabled()) log().debug("DataSender: category \'" + catlabel + "\' has listeners - converting to xml..."); int oldPriority = setCurrentThreadPriority(Thread.MAX_PRIORITY); EuiLevel euidata = null; try { euidata = m_euiMapper.convertToEuiLevelXML(cat); } catch (java.lang.OutOfMemoryError oe) { log().warn("DataSender: unable to convert data to xml for category: " + catlabel, oe); setCurrentThreadPriority(Thread.NORM_PRIORITY); continue; } catch (Throwable t) { log().warn("DataSender: unable to convert data to xml for category: " + catlabel, t); setCurrentThreadPriority(Thread.NORM_PRIORITY); } if (urlList != null && urlList.size() > 0) { Iterator<HttpPostInfo> urlIter = urlList.iterator(); while (urlIter.hasNext()) { HttpPostInfo postInfo = urlIter.next(); Reader inr = null; InputStream inp = null; try { inr = new PipedMarshaller(euidata).getReader(); if (log().isDebugEnabled()) log().debug("DataSender: posting data to: " + postInfo.getURLString()); inp = HttpUtils.post(postInfo.getURL(), inr, postInfo.getUser(), postInfo.getPassword(), 8 * HttpUtils.DEFAULT_POST_BUFFER_SIZE); if (log().isDebugEnabled()) log().debug("DataSender: posted data for category: " + catlabel); byte[] tmp = new byte[1024]; int bytesRead; while ((bytesRead = inp.read(tmp)) != -1) { if (log().isDebugEnabled()) { if (bytesRead > 0) log().debug("DataSender: post response: " + new String(tmp, 0, bytesRead)); } } postInfo.clearErrors(); } catch (IOException e) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + e.getClass().getName() + ": " + e.getMessage(), e); postInfo.incrementErrors(); setCurrentThreadPriority(Thread.NORM_PRIORITY); } catch (java.lang.OutOfMemoryError e) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + e.getClass().getName() + ": " + e.getMessage(), e); setCurrentThreadPriority(Thread.NORM_PRIORITY); } catch (RuntimeException e) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + e.getClass().getName() + ": " + e.getMessage(), e); setCurrentThreadPriority(Thread.NORM_PRIORITY); } catch (Throwable t) { log().warn("DataSender: unable to send data for category: " + catlabel + " due to " + t.getClass().getName() + ": " + t.getMessage(), t); setCurrentThreadPriority(Thread.NORM_PRIORITY); } finally { IOUtils.closeQuietly(inp); IOUtils.closeQuietly(inr); } if (POST_ERROR_LIMIT > 0 && postInfo.getErrors() >= POST_ERROR_LIMIT) { urlIter.remove(); log().warn("URL " + postInfo.getURLString() + " UNSUBSCRIBED due to reaching error limit " + postInfo.getErrors()); } } } oldPriority = setCurrentThreadPriority(oldPriority); } } DataSender(Map<String, RTCCategory> categories, int numSenders); synchronized void start(); synchronized void stop(); String getName(); int getStatus(); synchronized void subscribe(String url, String catlabel, String user, String passwd); synchronized void unsubscribe(final String urlStr); synchronized void sendData(); synchronized void notifyToSend();  }