/*
(non-Javadoc)
@see org.fcrepo.server.management.FedoraAPIMMTOM#getDatastreams(java.lang
.String pid ,)String asOfDateTime ,)String dsState )*
181745
@Override
    public List<org.fcrepo.server.types.gen.Datastream> getDatastreams(String pid,
                                                                       String asOfDateTime,
                                                                       String dsState) {
        assertInitialized();
        try {
            MessageContext ctx = context.getMessageContext();
            org.fcrepo.server.storage.types.Datastream[] intDatastreams =
                    m_management.getDatastreams(ReadOnlyContext
                            .getSoapContext(ctx), pid, DateUtility
                            .parseDateOrNull(asOfDateTime), dsState);
            return getGenDatastreams(intDatastreams);
        } catch (Throwable th) {
            LOG.error("Error getting datastreams", th);
            throw CXFUtility.getFault(th);
        }
    }
1352405
private static Property[] getDatastreamHeaders(String pid, Datastream ds) {
        Property[] result = new Property[3];
        result[0] = new Property(HttpHeaders.ACCEPT_RANGES,"bytes");
        result[1] = new Property(HttpHeaders.ETAG, Datastream.defaultETag(pid, ds));
        result[2] = new Property(HttpHeaders.LAST_MODIFIED, DateUtil.formatDate(ds.DSCreateDT));
        return result;
    }
459444
@Override
  public RecordStreamWithMetadata<D, S> processStream(RecordStreamWithMetadata<D, S> inputStream, WorkUnitState state) {
    Flowable<StreamEntity<D>> filteredStream =
        inputStream.getRecordStream().filter(r -> {
          if (r instanceof ControlMessage) {
            getMessageHandler().handleMessage((ControlMessage) r);
            return true;
          } else if (r instanceof RecordEnvelope) {
            boolean accept = executePolicies(((RecordEnvelope) r).getRecord(), this.results);
            if (!accept) {
              r.ack();
            }
            return accept;
          } else {
            return true;
          }
        });
    filteredStream = filteredStream.doFinally(this::close);
    return inputStream.withRecordStream(filteredStream);
  }
1012759
@Override
    public Set<Stream> getResultStreams() {
    	Set<Stream> streams = ImmutableSet.of(this.resultStream);
		return streams;
    }
702190
def setDatastreamState(self, pid, dsID, dsState):
        '''
        '''
        # /objects/{pid}/datastreams/{dsID} ? [dsState]
        http_args = {'dsState' : dsState}
        url = 'objects/%(pid)s/datastreams/%(dsid)s' % {'pid': pid, 'dsid': dsID}
        response = self.put(url, params=http_args)
        # returns response code 200 on success
        return response.status_code == requests.codes.ok
968178
@Override
    public org.fcrepo.server.types.mtom.gen.MIMETypedStream getDissemination(String pid,
                                                                             String serviceDefinitionPid,
                                                                             String methodName,
                                                                             org.fcrepo.server.types.mtom.gen.GetDissemination.Parameters parameters,
                                                                             String asOfDateTime) {
        MessageContext ctx = context.getMessageContext();
        Context context = ReadOnlyContext.getSoapContext(ctx);
        assertInitialized();
        try {
            org.fcrepo.server.storage.types.Property[] properties =
                    TypeUtility
                            .convertGenPropertyArrayToPropertyArray(parameters);
            org.fcrepo.server.storage.types.MIMETypedStream mimeTypedStream =
                    m_access.getDissemination(context,
                                              pid,
                                              serviceDefinitionPid,
                                              methodName,
                                              properties,
                                              DateUtility
                                                      .parseDateOrNull(asOfDateTime));
            org.fcrepo.server.types.mtom.gen.MIMETypedStream genMIMETypedStream =
                    TypeUtility
                            .convertMIMETypedStreamToGenMIMETypedStreamMTOM(mimeTypedStream);
            return genMIMETypedStream;
        } catch (Throwable th) {
            LOG.error("Error getting dissemination", th);
            throw CXFUtility.getFault(th);
        }
    }
996437
public byte[] getData(byte[] dst, int offset)
	{
		final int end = Math.min(data.length, dst.length - offset) / 3 * 3;
		for (int i = 0; i < end; ++i)
			dst[offset + i] = (byte) data[i];
		return dst;
	}

	/* (non-Javadoc)
	 * @see tuwien.auto.calimero.dptxlator.DPTXlator#getSubTypes()
	 */
	public Map getSubTypes()
	{
		return types;
	}

	/**
	 * @return the subtypes of the date translator type
	 * @see DPTXlator#getSubTypesStatic()
	 */
	protected static Map getSubTypesStatic()
	{
		return types;
	}

	private String fromDPT(int index)
	{
		if (sdf != null)
			synchronized (DPTXlatorDate.class) {
				try {
					return sdf.format(new Date(fromDPTMillis(index)));
				}
				catch (final KNXFormatException e) {}
				return "invalid date";
			}
		final int i = index * 3;
		// return year-month-day
		final int d = data[i + DAY];
		final int m = data[i + MONTH];
		return Short.toString(absYear(data[i + YEAR])) + '-' + align(m) + m + '-'
			+ align(d) + d;
	}

	private long fromDPTMillis(int index) throws KNXFormatException
	{
		synchronized (DPTXlatorDate.class) {
			getCalendar().clear();
			final int i = index * 3;
			c.set(Calendar.DAY_OF_MONTH, data[i + DAY]);
			c.set(Calendar.MONTH, data[i + MONTH] - 1);
			c.set(Calendar.YEAR, absYear(data[i + YEAR]));
			try {
				return c.getTimeInMillis();
			}
			catch (final IllegalArgumentException e) {
				throw new KNXFormatException(e.getMessage());
			}
		}
	}

	protected void toDPT(String value, short[] dst, int index) throws KNXFormatException
	{
		if (sdf != null)
			synchronized (DPTXlatorDate.class) {
				toDPT(parse(value), dst, index);
				return;
			}
		final StringTokenizer t = new StringTokenizer(value, "- ");
		final int maxTokens = 3;
		final int[] tokens = new int[maxTokens];
		try {
			int count = 0;
			for (; count < maxTokens && t.hasMoreTokens(); ++count)
				tokens[count] = Short.parseShort(t.nextToken());
			if (count < 3)
				throw logThrow(LogLevel.WARN, "invalid date " + value, null, value);
			set(tokens[0], tokens[1], tokens[2], dst, index);
		}
		catch (final KNXIllegalArgumentException e) {
			throw logThrow(LogLevel.WARN, "invalid date " + value, e.getMessage(), value);
		}
		catch (final NumberFormatException e) {
			throw logThrow(LogLevel.WARN, "invalid number in " + value, null, value);
		}
	}

	private short[] toDPT(long milliseconds, short[] dst, int index)
	{
		synchronized (DPTXlatorDate.class) {
			getCalendar().clear();
			c.setTimeInMillis(milliseconds);
			set(c.get(Calendar.YEAR), c.get(Calendar.MONTH) + 1, c
				.get(Calendar.DAY_OF_MONTH), dst, index);
		}
		return dst;
	}

	private short[] set(int year, int month, int day, short[] dst, int index)
	{
		if (year < 1990 || year > 2089)
			throw new KNXIllegalArgumentException("year out of range [1990..2089]");
		if (month < 1 || month > 12)
			throw new KNXIllegalArgumentException("month out of range [1..12]");
		if (day < 1 || day > 31)
			throw new KNXIllegalArgumentException("day out of range [1..31]");
		final int i = 3 * index;
		dst[i + DAY] = (short) day;
		dst[i + MONTH] = (short) month;
		dst[i + YEAR] = (short) (year % 100);
		return dst;
	}

	private short absYear(int relative)
	{
		if (relative > 99)
			throw new KNXIllegalArgumentException("relative year out of range [0..99]");
		return (short) (relative + (relative < 90 ? 2000 : 1900));
	}

	private long parse(String value) throws KNXFormatException
	{
		try {
			return sdf.parse(value).getTime();
		}
		catch (final ParseException e) {
			throw logThrow(LogLevel.WARN, "invalid date format", e.getMessage(), value);
		}
	}

	private static String align(int number)
	{
		return number > 9 ? "" : "0";
	}

	private static Calendar getCalendar()
	{
		// don't need to synchronize, it's harmless if we have 2 instances
		// and we synchronize on class anyway
		if (c == null) {
			final Calendar calendar = Calendar.getInstance();
			calendar.setLenient(false);
			c = calendar;
		}
		return c;
	}
}
779993
@Path("/{dsID}")
    @DELETE
    @Produces(MediaType.APPLICATION_JSON)
    public Response deleteDatastream(@PathParam(RestParam.PID) String pid,
                                     @PathParam(RestParam.DSID) String dsID,
                                     @QueryParam(RestParam.START_DT) String startDT,
                                     @QueryParam(RestParam.END_DT) String endDT,
                                     @QueryParam(RestParam.LOG_MESSAGE) String logMessage,
                                     @QueryParam(RestParam.FLASH) @DefaultValue("false") boolean flash) {

        try {
            Context context = getContext();
            Date startDate = DateUtility.parseDateOrNull(startDT);
            Date endDate = DateUtility.parseDateOrNull(endDT);
            Date[] purged = m_management.purgeDatastream(context,
                                        pid,
                                        dsID,
                                        startDate,
                                        endDate,
                                        logMessage);

            // convert purged into a String[] so we can return it as a JSON array
            List<String> results = new ArrayList<String>(purged.length);
            for (Date d : purged) {
                results.add(DateUtility.convertDateToXSDString(d));
            }
            return Response.ok(m_mapper.writeValueAsString(results)).build();
        } catch (Exception ex) {
            return handleException(ex, flash);
        }
    }
1167676
@Override
    protected BaasStream getFromCache(BaasBox box) throws BaasException {
        return box.mCache.getStream(id);
    }
735286
public Date setDatastreamVersionable(Context context,
                                         String pid,
                                         String dsID,
                                         boolean versionable,
                                         String logMessage)
            throws ServerException {
        return worker.setDatastreamVersionable(context,
                                               pid,
                                               dsID,
                                               versionable,
                                               logMessage);
    }
342561
private static DatastreamInfo convertGenDatastreamDefToDatastreamInfo(Datastream ds) {
        return new DatastreamInfo(ds.getID(), ds.getMIMEType(), ds
                .getFormatURI());
    }
1333064
def datastream_lastmodified(request, pid, dsid, repo=None,
    as_of_date=None, *args, **kwargs):
    '''
    '''
    try:
        if repo is None:
            repo = Repository()
        resp = repo.api.getDatastream(pid, dsid, asOfDateTime=as_of_date)
        dsprofile = parse_xml_object(DatastreamProfile, resp.content, resp.url)
        if dsprofile:
            return dsprofile.created
    except RequestFailed:
        pass
14285
public byte[] getData(byte[] dst, int offset)
	{
		final int end = Math.min(data.length, dst.length - offset) / 3 * 3;
		for (int i = 0; i < end; ++i)
			dst[offset + i] = (byte) data[i];
		return dst;
	}

	/* (non-Javadoc)
	 * @see tuwien.auto.calimero.dptxlator.DPTXlator#getSubTypes()
	 */
	public Map getSubTypes()
	{
		return types;
	}

	/**
	 * @return the subtypes of the time translator type
	 * @see DPTXlator#getSubTypesStatic()
	 */
	protected static Map getSubTypesStatic()
	{
		return types;
	}

	private String fromDPT(int index)
	{
		if (sdf != null)
			synchronized (DPTXlatorTime.class) {
				return sdf.format(new Date(fromDPTMillis(index)));
			}
		final int i = index * 3;
		// dow, hh:mm:ss
		final int h = data[i + HOUR] & 0x1F;
		final int m = data[i + MINUTE];
		final int s = data[i + SECOND];
		return DAYS[data[i + DOW] >> 5] + ", " + align(h) + h + ':' + align(m) + m + ':'
			+ align(s) + s;
	}

	private long fromDPTMillis(int index)
	{
		synchronized (DPTXlatorTime.class) {
			getCalendar().clear();
			final int i = index * 3;
			final int dow = data[i] >> 5 & 0x7;
			if (dow > 0)
				c.set(Calendar.DAY_OF_WEEK, dow == 7 ? 1 : dow + 1);
			c.set(Calendar.HOUR_OF_DAY, data[i + HOUR] & 0x1F);
			c.set(Calendar.MINUTE, data[i + MINUTE]);
			c.set(Calendar.SECOND, data[i + SECOND]);
			long ms = 0;
			try {
				ms = c.getTimeInMillis();
			}
			catch (final IllegalArgumentException e) {
				DPTXlator.logger.error("from DPT in milliseconds not possible");
			}
			return ms;
		}
	}

	protected void toDPT(String value, short[] dst, int index) throws KNXFormatException
	{
		if (sdf != null)
			synchronized (DPTXlatorTime.class) {
				toDPT(parse(value), dst, index);
				return;
			}
		final StringTokenizer t = new StringTokenizer(value, ": ,");
		final int maxTokens = 4;
		final String[] tokens = new String[maxTokens];
		// extract tokens
		int count = 0;
		for (; count < maxTokens && t.hasMoreTokens(); ++count)
			tokens[count] = t.nextToken();
		// we allow day of week to be omitted in value
		if (count < 3)
			throw logThrow(LogLevel.WARN, "invalid time " + value, null, value);
		// on 4 tokens, day of week is included
		final int dow = count == 4 ? getDOW(tokens[0]) : 0;
		try {
			final int s = Short.parseShort(tokens[--count]);
			final int m = Short.parseShort(tokens[--count]);
			final int h = Short.parseShort(tokens[--count]);
			set(dow, h, m, s, dst, index);
		}
		catch (final KNXIllegalArgumentException e) {
			throw logThrow(LogLevel.WARN, "invalid time " + value, e.getMessage(), value);
		}
		catch (final NumberFormatException e) {
			throw logThrow(LogLevel.WARN, "invalid number in " + value, null, value);
		}
	}

	private short[] toDPT(long milliseconds, short[] dst, int index)
	{
		synchronized (DPTXlatorTime.class) {
			getCalendar().clear();
			c.setTimeInMillis(milliseconds);
			final int dow = c.get(Calendar.DAY_OF_WEEK);
			set(dow == Calendar.SUNDAY ? 7 : dow - 1, c
				.get(Calendar.HOUR_OF_DAY), c.get(Calendar.MINUTE), c
				.get(Calendar.SECOND), dst, index);
		}
		return dst;
	}

	private short[] set(int dow, int hour, int minute, int second, short[] dst, int index)
	{
		if (dow < 0 || dow > 7)
			throw new KNXIllegalArgumentException("day of week out of range [0..7]");
		if (hour < 0 || hour > 23)
			throw new KNXIllegalArgumentException("hour out of range [0..23]");
		if (minute < 0 || minute > 59)
			throw new KNXIllegalArgumentException("minute out of range [0..59]");
		if (second < 0 || second > 59)
			throw new KNXIllegalArgumentException("second out of range [0..59]");
		final int i = 3 * index;
		dst[i + DOW] = (short) (dow << 5 | hour);
		dst[i + MINUTE] = (short) minute;
		dst[i + SECOND] = (short) second;
		return dst;
	}

	private int getDOW(String dow) throws KNXFormatException
	{
		for (int i = 0; i < DAYS.length; ++i)
			if (DAYS[i].equalsIgnoreCase(dow))
				return i;
		throw logThrow(LogLevel.WARN, "invalid day of week " + dow, null, dow);
	}

	private long parse(String value) throws KNXFormatException
	{
		try {
			return sdf.parse(value).getTime();
		}
		catch (final ParseException e) {
			throw logThrow(LogLevel.WARN, "invalid time format", e.getMessage(), value);
		}
	}

	private static String align(int number)
	{
		return number > 9 ? "" : "0";
	}

	private static Calendar getCalendar()
	{
		// don't need to synchronize, it's harmless if we have 2 instances
		// and we synchronize on class anyway
		if (c == null) {
			final Calendar calendar = Calendar.getInstance();
			calendar.setLenient(false);
			c = calendar;
		}
		return c;
	}
}
747884
public Date[] purgeDatastream(Context context,
                                  String pid,
                                  String datastreamID,
                                  Date startDT,
                                  Date endDT,
                                  String logMessage) throws ServerException {
        try {
            CreatorJournalEntry cje =
                    new CreatorJournalEntry(METHOD_PURGE_DATASTREAM, context);
            cje.addArgument(ARGUMENT_NAME_PID, pid);
            cje.addArgument(ARGUMENT_NAME_DS_ID, datastreamID);
            cje.addArgument(ARGUMENT_NAME_START_DATE, startDT);
            cje.addArgument(ARGUMENT_NAME_END_DATE, endDT);
            cje.addArgument(ARGUMENT_NAME_LOG_MESSAGE, logMessage);
            return (Date[]) cje.invokeAndClose(delegate, writer);
        } catch (JournalException e) {
            throw new GeneralException("Problem creating the Journal", e);
        }
    }
1104873
@Override
	public DataStream<Row> getDataStream(StreamExecutionEnvironment env) {

		DeserializationSchema<Row> deserializationSchema = getDeserializationSchema();
		// Version-specific Kafka consumer
		FlinkKafkaConsumerBase<Row> kafkaConsumer = getKafkaConsumer(topic, properties, deserializationSchema);
		return env.addSource(kafkaConsumer).name(explainSource());
	}
379909
public Date setDatastreamState(Context context,
                                   String pid,
                                   String dsID,
                                   String dsState,
                                   String logMessage) throws ServerException {
        try {
            CreatorJournalEntry cje =
                    new CreatorJournalEntry(METHOD_SET_DATASTREAM_STATE,
                                            context);
            cje.addArgument(ARGUMENT_NAME_PID, pid);
            cje.addArgument(ARGUMENT_NAME_DS_ID, dsID);
            cje.addArgument(ARGUMENT_NAME_DS_STATE, dsState);
            cje.addArgument(ARGUMENT_NAME_LOG_MESSAGE, logMessage);
            return (Date) cje.invokeAndClose(delegate, writer);
        } catch (JournalException e) {
            throw new GeneralException("Problem creating the Journal", e);
        }
    }
964408
public Date[] purgeDatastream(Context context,
                                  String pid,
                                  String datastreamID,
                                  Date startDT,
                                  Date endDT,
                                  String logMessage) throws ServerException {
        return worker.purgeDatastream(context,
                                      pid,
                                      datastreamID,
                                      startDT,
                                      endDT,
                                      logMessage);
    }