import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
/**
 * Handle a String message, returning a list of AisPackets
 * 
 * @param line
 * @return
 */
public List<AisPacket> handleLine(String messageString) throws IOException {
	if (logger.isDebugEnabled())
		logger.debug("AIS Received : " + messageString);
	// Check for ABK
	if (Abk.isAbk(messageString)) {
		if (logger.isDebugEnabled())
			logger.debug("AIS Received ABK: " + messageString);
		return null;
	}

	try {
		List<AisPacket> packets = new ArrayList<AisPacket>();
		String[] lines = messageString.split("\\r?\\n");

		for (String line : lines) {
			packets.add(packetParser.readLine(line));
		}
		return packets;
	} catch (SentenceException se) {
		if (logger.isDebugEnabled())
			logger.info("AIS Sentence error: " + se.getMessage() + " line: " + messageString);
		throw new IOException(se);

	}
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
private void startAisBus(Consumer<AisPacket> packetConsumer) {
    LOG.debug("Starting AisBus");
    try {
        DistributerConsumer distributor = new DistributerConsumer();
        distributor.getConsumers().add(packetConsumer);
        distributor.init();
        aisBus.registerConsumer(distributor);
        aisBus.start();
        aisBus.startConsumers();
        aisBus.startProviders();
    } catch (Exception e) {
        LOG.error("Failed to start AisBus", e);
        exit(-1);
    }
    LOG.debug("AisBus started");
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
public void receiveUnfiltered(AisPacket packet) {
    unfiltCount++;
    
    //extract relevant information from packet
    CustomMessage message = dataHandler.packetToCustomMessage(packet);
    if(message == null){
        return;
    }
    
    String key = message.getKey();
    
    //Add to doublet buffer.
    CustomMessage existing = doubletBuffer.get(key);
    if (existing == null) {
        doubletBuffer.put(key, message);
    }else{
        existing.addSourceMMSI(message.getSourceList().iterator().next());
    }

}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
@GET
@Path("/static/{mmsi : \\d+}")
@Produces(MediaType.TEXT_PLAIN)
public StreamingOutput getStaticInfo(@Context UriInfo info,
        @PathParam("mmsi") int mmsi) {
    QueryParameterHelper qh = new QueryParameterHelper(info);

    Predicate<AisPacketSource> pred = qh.getSourcePredicate();
    pred = (pred == null) ? e -> true : pred;

    TargetTracker tt = get(TargetTracker.class);
    TargetInfo ti = tt.get(mmsi, pred);
    List<AisPacket> l = java.util.Arrays.asList(ti.getStaticPackets());

    Iterable<AisPacket> p = applyFilters(l.stream(), qh);
    
    return StreamingUtil.createStreamingOutput(p,
            qh.getOutputSink());
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
@GET
@Path("/dynamic/{mmsi : \\d+}")
@Produces(MediaType.TEXT_PLAIN)
public StreamingOutput getPositionInfo(@Context UriInfo info,
        @PathParam("mmsi") int mmsi) {
    QueryParameterHelper qh = new QueryParameterHelper(info);

    Predicate<AisPacketSource> pred = qh.getSourcePredicate();
    pred = (pred == null) ? e -> true : pred;

    TargetTracker tt = get(TargetTracker.class);
    TargetInfo ti = tt.get(mmsi, pred);

    // convert TargetInfo to Stream<AisPacket> and then Iterable<AisPacket>
    // after applying filters
    Iterable<AisPacket> packets = applyFilters(
            Arrays.asList(new TargetInfo[] { ti }).stream()
                    .map(e -> e.getPositionPacket()), qh);

    return StreamingUtil.createStreamingOutput(packets,
            qh.getOutputSink());
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
@GET
@Path("/dynamic/")
@Produces(MediaType.TEXT_PLAIN)
public StreamingOutput getPositions(@Context UriInfo info) {
    QueryParameterHelper qh = new QueryParameterHelper(info);
    Predicate<AisPacketSource> predSource = qh.getSourcePredicate();
    predSource = (predSource == null) ? e -> true : predSource;

    Predicate<TargetInfo> predTarget = qh.getTargetPredicate();
    predTarget = predTarget.and(TargetInfoFilters.filterOnHasPosition());
    predTarget = (qh.getArea() != null) ? qh.getTargetAreaFilter() : predTarget;

    TargetTracker tt = get(TargetTracker.class);

    Stream<TargetInfo> s = tt.stream(predSource, predTarget);
    Stream<AisPacket> sPacket = s.map(e -> e.getPositionPacket()).filter(
            o -> o != null);

    Iterable<AisPacket> packets = applyFilters(sPacket, qh);

    
    
    return StreamingUtil.createStreamingOutput(packets,
            qh.getOutputSink());
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
protected Stream<AisPacket> getPacketStream(UriInfo info, QueryParameterHelper qh) {
    Predicate<AisPacketSource> predSource = qh.getSourcePredicate();
    predSource = (predSource == null) ? e -> true : predSource;

    Predicate<TargetInfo> predTarget = qh.getTargetPredicate();
    predTarget = (predTarget == null) ? e -> true : predTarget;

    predTarget = (qh.getArea() != null) ? qh.getTargetAreaFilter() : predTarget;

    TargetTracker tt = get(TargetTracker.class);

    Stream<TargetInfo> s = tt.stream(predSource, predTarget);
    Stream<AisPacket[]> sPackets = s.map(e -> e.getPackets()).filter(o -> o != null);

    final ConcurrentLinkedDeque<AisPacket> packets = new ConcurrentLinkedDeque<AisPacket>();
    sPackets.sequential().forEach(e -> {
        for (int i = 0; i < e.length; i++) {
            packets.add(e[i]);
        }
    });

    return packets.stream();
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
/**
 * Get the count of messages from packets received with timestamp in the
 * closest full 10 minutes time block.
 * 
 * @return
 */
private AtomicLong getTenMinuteCount() {
    // get the latest guaranteed full block
    long startBlock = (long) ((double) DateTime.now().getMillis() / 10.0 / 60.0 / 1000.0) - 1;
    long endBlock = startBlock + 1;

    long start = startBlock * 10 * 60 * 1000;
    long end = endBlock * 10 * 60 * 1000;

    AisStoreQueryBuilder b = AisStoreQueryBuilder.forTime().setInterval(
            start, end);
    AisStoreQueryResult query = cassandraConnection().execute(b);
    Iterable<AisPacket> q = query;

    final AtomicLong l = new AtomicLong();
    q = Iterables.counting(q, l);
    for (Iterator<AisPacket> iterator = q.iterator(); iterator.hasNext();) {
        iterator.next();
    }

    return l;
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
/**
 * getPastTrack will only work with position messages
 * 
 * @param info
 * @param mmsi
 * @return
 */
private Iterable<AisPacket> getPastTrack(@Context UriInfo info, int... mmsi) {
    QueryParameterHelper p = new QueryParameterHelper(info);

    // Execute the query
    AisStoreQueryBuilder b = AisStoreQueryBuilder.forMmsi(mmsi);
    b.setInterval(p.getInterval());

    // Create the query
    Iterable<AisPacket> query = cassandraConnection().execute(b);

    // Apply filters from the user
    query = Iterables.filter(query, AisPacketFilters
            .filterOnMessageType(IVesselPositionMessage.class));
    query = p.applyPacketFilter(query);
    query = p.applyPositionSampler(query); // WARNING: Must be the second
                                           // last filter
    query = p.applyLimitFilter(query); // WARNING: Must be the last filter
                                       // (if other filters reject packets)
    return query;
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
/**
 * getHistory takes all ais data with mmsis using stateful filters
 * 
 * @param info
 * @param mmsi
 * @return
 */
private Iterable<AisPacket> getHistory(@Context UriInfo info, int... mmsi) {
    QueryParameterHelper p = new QueryParameterHelper(info);

    // Execute the query
    AisStoreQueryBuilder b = AisStoreQueryBuilder.forMmsi(mmsi);
    b.setInterval(p.getInterval());

    // Create the query
    Iterable<AisPacket> query = cassandraConnection().execute(b);

    final AisPacketFiltersStateful state = new AisPacketFiltersStateful();

    // Apply filters from the user
    query = p.applyPacketFilter(query); // first because this potentially
                                        // filters a lot of packets
    query = p.applyTargetFilterArea(query, state);
    query = p.applyLimitFilter(query); // WARNING: Must be the last filter
                                       // (if other filters reject packets)
    return query;
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
@Provides
@Named("shipNameFilter")
Predicate<AisPacket> provideShipNameFilter() {
    Configuration configuration = getConfiguration();
    String[] shipNames = configuration.getStringArray(CONFKEY_FILTER_SHIPNAME_SKIP);
    Predicate<AisPacket> filter = null;
    if (shipNames != null && shipNames.length > 0 && !(shipNames.length == 1 && shipNames[0].trim().length() == 0)) {
        String filterExpression = "";
        for (int i = 0; i < shipNames.length; i++) {
            filterExpression += "t.name ~ " + shipNames[i];
            if (i != shipNames.length - 1) {
                filterExpression += " | ";
            }
        }
        LOG.debug("filterExpression: " + filterExpression);
        filter = parseExpressionFilter(filterExpression);
        LOG.info("Created ship name filter: " + filterExpression);
    }
    return filter != null ? filter : aisPacket -> false;
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
@Inject
public PacketHandlerImpl(
        Configuration configuration,
        Injector injector,
        AppStatisticsService statisticsService,
        EventEmittingTracker tracker,
        Set<IPacketFilter> filters,
        @Named("shipNameFilter") Predicate<AisPacket> shipNameFilter
) {
    this.configuration = configuration;
    this.injector = injector;
    this.statisticsService = statisticsService;
    this.tracker = tracker;
    this.filters = filters;
    this.shipNameFilter = shipNameFilter;
    this.analyses = initAnalyses();

    this.analyses.forEach(analysis -> analysis.start());
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
/**
 * Receive and process one AisPacket.
 *
 * @param packet The AisPacket to process.
 */
public void accept(final AisPacket packet) {
    statisticsService.incUnfilteredPacketCount();
    if (filterPacket(packet)) {
        statisticsService.incFilteredPacketCount();

        AisMessage message = packet.tryGetAisMessage();
        if (message == null) {
            LOG.warn("Invalid packet: " + packet.getStringMessage());
            return;
        }
        updateApplicationStatistics(message);

        doWork(packet);
    }
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
/**
 * Returns true if packet passes all packet filters
 * @param packet
 * @return
 */
private boolean filterPacket(AisPacket packet) {
    final boolean[] rejected = {false}; // TODO Use Groovy or Scala...
    filters.forEach(f -> {
        if (!rejected[0] && f.rejectedByFilter(packet)) {
            rejected[0] = true;
            if (LOG.isDebugEnabled()) {
                LOG.debug("Packet dropped due to " + f.getClass().getSimpleName());
            }
        }
    });

    boolean shipNameFilterPassed = ! shipNameFilter.test(packet);
    if (!shipNameFilterPassed && LOG.isDebugEnabled()) {
        LOG.debug("Packet dropped due to shipNameFilter");
    }

    boolean filterPassed = !rejected[0] && shipNameFilterPassed;

    if (LOG.isDebugEnabled()) {
        LOG.debug("Packet " + (filterPassed ? "passed":"dropped") + ": " + packet.getStringMessage());
    }

    return filterPassed;
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
private AisCoverage(AisCoverageConfiguration conf) {
    this.conf = conf;

    // Create handler
    handler = new CoverageHandler(conf);

    // Create AisBus
    aisBus = conf.getAisbusConfiguration().getInstance();
    

    // Create web server
    if (conf.getServerConfiguration() != null) {
        webServer = new WebServer(conf.getServerConfiguration());
    } else {
        webServer = null;
    }

    final DistributerConsumer unfilteredConsumer = new DistributerConsumer(true);
    unfilteredConsumer.init();
    // Delegate unfiltered packets to handler
    unfilteredConsumer.getConsumers().add(new Consumer<AisPacket>() {
        @Override
        public void accept(AisPacket packet) {
            handler.receiveUnfiltered(packet);
        }
    });                
    aisBus.registerConsumer(unfilteredConsumer);
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
public AisTargetEntry(AisPacket packet) {
    AisMessage aisMessage = packet.tryGetAisMessage();        
    this.anonId = ++anonymousCounter;
    mmsiAnonIdMap.put(aisMessage.getUserId(), this.anonId);
    anonIdMap.put(this.anonId, aisMessage.getUserId());
    this.target = AisTarget.createTarget(aisMessage);
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
public boolean update(AisPacket packet) {
    boolean targetReplaced = false;
    AisMessage aisMessage = packet.tryGetAisMessage();
    sourceData.update(packet);
    // We want to avoid to update a target position with an older position
    // than the last one received
    boolean oldPos = false;
    if (aisMessage instanceof IVesselPositionMessage) {
        Date thisReport = null;
        // Get timestamp for message tag or fallback to time now
        thisReport = aisMessage.getVdm().getTimestamp();
        if (thisReport == null) {
            thisReport = new Date();
        }
        if (lastReport != null) {
            // We will not update if this report is older than last
            if (thisReport.before(lastReport)) {
                oldPos = true;
            }
        }
        lastReport = thisReport;
    }
    // Update target data
    if (!oldPos) {
        try {
            target.update(aisMessage);
        } catch (IllegalArgumentException e) {
            // Trying to update target with report of different type of target.
            // Replace target with new target
            target = AisTarget.createTarget(aisMessage);
            target.update(aisMessage);
            targetReplaced = true;
        }
    }

    return targetReplaced;
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
@GET
@Path("rate/count")
@Produces(MediaType.TEXT_PLAIN)
public Long rateCount(@Context UriInfo uriInfo) {
    TargetTracker tt = Objects.requireNonNull(LegacyResource.this
            .get(TargetTracker.class));

    Stream<TargetInfo> tis = tt.stream();

    final Date d = new Date(new Date().getTime() - 1000);
    final Long dLong = d.getTime();

    // count all packets in the TargetInfo which are after the given
    // timestamp.
    // this is kind of "double work" since we already filtered them above.
    Long r = tis.mapToLong(value -> {

        long r1 = 0L;
        if (value.getPositionTimestamp() > dLong) {
            r1++;
        }

        // eep, this is slow but has to be done
        if (value.getStaticCount() > 1) {
            for (AisPacket p : value.getStaticPackets()) {
                if (p.getBestTimestamp() > dLong) {
                    r1++;
                }
            }
        } else if (value.getStaticTimestamp() > dLong) {
            r1++;
        }

        return r1;
    }).sum();

    return new Double((double) r).longValue();
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
@GET
@Path("/static/")
@Produces(MediaType.TEXT_PLAIN)
public StreamingOutput getStatics(@Context UriInfo info) {
    QueryParameterHelper qh = new QueryParameterHelper(info);

    Predicate<AisPacketSource> predSource = qh.getSourcePredicate();
    predSource = (predSource == null) ? e -> true : predSource;

    Predicate<TargetInfo> predTarget = qh.getTargetPredicate();
    predTarget = (predTarget == null) ? e -> true : predTarget;
    
    predTarget.and(TargetInfoFilters.filterOnHasStatic());
    predTarget = (qh.getArea() != null) ? qh.getTargetAreaFilter() : predTarget;

    TargetTracker tt = get(TargetTracker.class);

    Stream<TargetInfo> s = tt.stream(predSource, predTarget);

    Stream<AisPacket[]> sPackets = s.map(e -> e.getStaticPackets());

    final ConcurrentLinkedDeque<AisPacket> packets = new ConcurrentLinkedDeque<AisPacket>();
    sPackets.forEach(e -> {
        for (int i = 0; i < e.length; i++) {
            packets.add(e[i]);
        }
    });

    Iterable<AisPacket> p = applyFilters(packets.stream(), qh);

    return StreamingUtil.createStreamingOutput(p,
            qh.getOutputSink());
}
 

import dk.dma.ais.packet.AisPacket; //导入依赖的package包/类
@GET
@Path("/packets/json")
@Produces(MediaType.TEXT_PLAIN)
public StreamingOutput getPacketsJson(@Context UriInfo info) {
    QueryParameterHelper qh = new QueryParameterHelper(info);
    Stream<AisPacket> packets = getPacketStream(info, qh);
    return StreamingUtil.createStreamingOutput(
            (Iterable<AisPacket>) packets::iterator,
            AisPacketOutputSinks.jsonMessageSink());
}