WowzaStreamingDistributionService extends AbstractDistributionService implements StreamingDistributionService { @Override public Job distribute(String channelId, MediaPackage mediapackage, Set<String> elementIds) throws DistributionException { notNull(mediapackage, "mediaPackage"); notNull(elementIds, "elementIds"); notNull(channelId, "channelId"); if (getStreamingURLforCurrentOrg() == null) throw new IllegalStateException( String.format("No streaming url or port set for tenant %s", securityService.getOrganization().getId())); if (distributionDirectory == null) throw new IllegalStateException( "Streaming distribution directory must be set (org.opencastproject.streaming.directory)"); try { return serviceRegistry.createJob( JOB_TYPE, Operation.Distribute.toString(), Arrays.asList(channelId, MediaPackageParser.getAsXml(mediapackage), gson.toJson(elementIds)), distributeJobLoad); } catch (ServiceRegistryException e) { throw new DistributionException("Unable to create a job", e); } } WowzaStreamingDistributionService(); String getDistributionType(); void activate(BundleContext bundleContext, Map<String, Object> properties); void modified(BundleContext bundleContext, Map<String, Object> properties); boolean publishToStreaming(); @Override Job distribute(String channelId, MediaPackage mediapackage, Set<String> elementIds); @Override Job distribute(final String channelId, final MediaPackage mediapackage, final String elementId); @Override Job retract(String channelId, MediaPackage mediapackage, String elementId); @Override Job retract(String channelId, MediaPackage mediaPackage, Set<String> elementIds); @Override List<MediaPackageElement> distributeSync(String channelId, MediaPackage mediaPackage, Set<String> elementIds); @Override List<MediaPackageElement> retractSync(String channelId, MediaPackage mediaPackage, Set<String> elementIds); static final String JOB_TYPE; static final float DEFAULT_DISTRIBUTE_JOB_LOAD; static final float DEFAULT_RETRACT_JOB_LOAD; static final String DISTRIBUTE_JOB_LOAD_KEY; static final String RETRACT_JOB_LOAD_KEY; }
@Test public void testDistribute() throws Exception { setUpMultiTenant(); expect(securityService.getOrganization()).andReturn(org1).anyTimes(); replay(securityService); streamingService.activate(bundleContext, map); MediaPackage mp = MediaPackageParser.getFromXml( IOUtils.toString(getClass().getResourceAsStream("/mediapackage.xml"), "UTF-8")); Job job = streamingService.distribute("channel", mp, "blah"); JobBarrier jobBarrier = new JobBarrier(null, serviceRegistry, 500, job); jobBarrier.waitForJobs(); }