import org.postgresql.PGNotification; //导入依赖的package包/类
@Test
public void testNotify() throws Exception {

    PGListener l = new PGListener(ds, bus, tester);

    when(conn.getNotifications(anyInt())).thenReturn(new PGNotification[] { //
            new Notification(PGConstants.CHANNEL_NAME, 1), //
            new Notification(PGConstants.CHANNEL_NAME, 1), //
            new Notification(PGConstants.CHANNEL_NAME, 1) }, //
            new PGNotification[] { new Notification(PGConstants.CHANNEL_NAME, 2) },
            new PGNotification[] { new Notification(PGConstants.CHANNEL_NAME, 3) }, //
            null);
    l.afterPropertiesSet();

    sleep(400);

    // 4 posts: one scheduled, 3 from notifications
    verify(bus, times(4)).post(any(FactInsertionEvent.class));

}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
public void run()
{
    trysleep(2000);
    while (!done) {
        try {
            PGConnection pg = (PGConnection)getConnection();
            if (pg == null) {
                log.info("watcher: connection was null");
                trysleep(5000);
                continue;
            }
            PGNotification notifications[] = pg.getNotifications();
            if (notifications != null) {
                Set<String> changes = new HashSet<String>();
                for (PGNotification n : notifications) {
                    changes.add(n.getParameter());
                }
                Messenger.sendEvent(MT.DATABASE_NOTIFICATION, changes);
            }
        } catch (Throwable e) {
            log.log(Level.WARNING, "ConnectionWatcher exception: " + e, e);
        }

        trysleep(500);
    }
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
/**
 * Parses base type {@link PGNotification notifications} for synchronizing the state of
 * {@link BaseType base types} between multiple instances of BinNavi.
 * 
 * @param notification THe {@link PGNotification} from the database.
 * @return A {@link TypesNotificationContainer container} with the parsed information.
 */
private TypesNotificationContainer parseBaseTypesNotification(final PGNotification notification) {

  final Matcher matcher = BASE_TYPES_NOTIFICATION_PATTERN.matcher(notification.getParameter());

  if (!matcher.find()) {
    throw new IllegalStateException("Error: compiled pattern: " + BASE_TYPE_NOTIFICATION_REGEX
        + " did not match notification: " + notification.getParameter());
  }

  final String databaseOperation = matcher.group(2);
  final Integer moduleId = Integer.parseInt(matcher.group(3));
  final Integer baseTypeId = Integer.parseInt(matcher.group(4));

  return new TypesNotificationContainer(databaseOperation, moduleId, Optional.of(baseTypeId),
      Optional.<Integer>absent(), Optional.<BigInteger>absent(), Optional.<Integer>absent(),
      Optional.<Integer>absent());
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
/**
 * Parses expression type aka type substitution {@link PGNotification notifications} for
 * synchronizing the state of {@link TypeSubstitution type substitutions} between multiple
 * instances of BinNavi.
 * 
 * @param notification The {@link PGNotification} from the database.
 * @return A {@link TypesNotificationContainer container} with the parsed information.
 */
private TypesNotificationContainer parseExpressionTypesNotification(
    final PGNotification notification) {

  final Matcher matcher =
      EXPRESSION_TYPES_NOTIFICATION_PATTERN.matcher(notification.getParameter());

  if (!matcher.find()) {
    throw new IllegalStateException("Error: compliled pattern "
        + EXPRESSION_TYPES_NOTIFICATION_PATTERN.toString() + " did not match notification: "
        + notification.getParameter());
  }

  final String databaseOperation = matcher.group(2);
  final Integer moduleId = Integer.parseInt(matcher.group(3));
  final BigInteger address = new BigInteger(matcher.group(4));
  final Integer position = Integer.parseInt(matcher.group(5));
  final Integer expressionId = Integer.parseInt(matcher.group(6));

  return new TypesNotificationContainer(databaseOperation, moduleId, Optional.<Integer>absent(),
      Optional.<Integer>absent(), Optional.of(address), Optional.of(position),
      Optional.of(expressionId));
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
/**
 * Parses type {@link PGNotification notifications} for synchronizing the state of
 * {@link TypeMember type members} between multiple instances of BinNavi.
 * 
 * @param notification The {@link PGNotification} from the database.
 * @return A {@link TypesNotificationContainer container} with the parsed information.
 */
private TypesNotificationContainer parseTypesNotification(final PGNotification notification) {

  final Matcher matcher = TYPES_NOTIFICATION_PATTERN.matcher(notification.getParameter());

  if (!matcher.find()) {
    throw new IllegalStateException("Error: compiled pattern: " + TYPES_NOTIFICATION_REGEX
        + " did not match notification: " + notification.getParameter());
  }

  final String databaseOperation = matcher.group(2);
  final Integer moduleId = Integer.parseInt(matcher.group(3));
  final Integer typeMemberId = Integer.parseInt(matcher.group(4));

  return new TypesNotificationContainer(databaseOperation, moduleId, Optional.<Integer>absent(),
      Optional.of(typeMemberId), Optional.<BigInteger>absent(), Optional.<Integer>absent(),
      Optional.<Integer>absent());

}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
@Override
public Collection<TypesNotificationContainer> parse(
    final Collection<PGNotification> notifications, final SQLProvider provider) {

  Preconditions.checkNotNull(notifications, "Error: notifications argument can not be null");
  Preconditions.checkNotNull(provider, "Error: provider argument can not be null");

  final Collection<TypesNotificationContainer> containers = Lists.newArrayList();

  for (final PGNotification notification : notifications) {
    if (notification.getParameter().startsWith(CTableNames.EXPRESSION_TYPES_TABLE)) {
      containers.add(parseExpressionTypesNotification(notification));
    } else if (notification.getParameter().startsWith(CTableNames.TYPE_MEMBERS_TABLE)) {
      containers.add(parseTypesNotification(notification));
    } else if (notification.getParameter().startsWith(CTableNames.BASE_TYPES_TABLE)) {
      containers.add(parseBaseTypesNotification(notification));
    } else {
      throw new IllegalStateException("Error: Table name supplied in notification "
          + notification.getParameter()
          + " does not match tables where type notifications are expected on.");
    }
  }
  return containers;
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
/**
 * Parses type instance notifications. This function parses messages for synchronization of
 * changes in the {@link CTableNames bn_type_instances} table.
 * 
 * @param notification The {@link PGNotification} which carries the information to be parsed.
 * @return A {@link TypeInstancesNotificationContainer} with the parsed information.
 */
private TypeInstancesNotificationContainer parseTypeInstanceNotification(
    final PGNotification notification) {

  final Matcher matcher = typeInstanceNotificationPattern.matcher(notification.getParameter());
  if (!matcher.find()) {
    throw new IllegalStateException("Error: compiled pattern: "
        + typeInstanceNotificationPattern.toString() + " did not match notification: "
        + notification.getParameter());
  }

  final String databaseOperation = matcher.group(2);
  final Integer moduleId = Integer.parseInt(matcher.group(3));
  final Integer typeInstanceId = Integer.parseInt(matcher.group(4));

  return new TypeInstancesNotificationContainer(databaseOperation, moduleId, typeInstanceId,
      Optional.<BigInteger>absent(), Optional.<Integer>absent(), Optional.<Integer>absent());
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
/**
 * Parses expression type instances or simpler type substitutions on operands. This function
 * parses messages for synchronization of changes in the {@link CTableNames
 * bn_expression_type_instances} table.
 * 
 * @param notification The {@link PGNotification} which carries the information to be parsed.
 * @return A {@link TypeInstancesNotificationContainer} with the parsed information.
 */
private TypeInstancesNotificationContainer parseExpressionTypeInstanceNotification(
    final PGNotification notification) {

  final Matcher matcher =
      expressionTypeInstanceNotificationPattern.matcher(notification.getParameter());
  if (!matcher.find()) {
    throw new IllegalStateException("Error: compiled pattern: "
        + expressionTypeInstanceNotificationPattern.toString() + " did not match notification: "
        + notification.getParameter());
  }

  final String databaseOperation = matcher.group(2);
  final Integer moduleId = Integer.parseInt(matcher.group(3));
  final BigInteger address = new BigInteger(matcher.group(4));
  final Integer position = Integer.parseInt(matcher.group(5));
  final Integer expressionId = Integer.parseInt(matcher.group(6));
  final Integer typeInstanceId = Integer.parseInt(matcher.group(7));

  return new TypeInstancesNotificationContainer(databaseOperation, moduleId, typeInstanceId,
      Optional.of(address), Optional.of(position), Optional.of(expressionId));
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
@Override
public Collection<TypeInstancesNotificationContainer> parse(
    final Collection<PGNotification> notifications, final SQLProvider provider) {

  Preconditions.checkNotNull(notifications, "Error: notifications argument can not be null");
  Preconditions.checkNotNull(provider, "Error: provider argument can not be null");

  final Collection<TypeInstancesNotificationContainer> containers = Lists.newArrayList();

  for (final PGNotification notification : notifications) {
    if (notification.getParameter().startsWith(CTableNames.TYPE_INSTANCE_TABLE)) {
      containers.add(parseTypeInstanceNotification(notification));
    } else if (notification.getParameter()
        .startsWith(CTableNames.EXPRESSION_TYPE_INSTANCES_TABLE)) {
      containers.add(parseExpressionTypeInstanceNotification(notification));
    } else {
      throw new IllegalStateException("Error: Table name supplied in notification "
          + notification.getParameter()
          + " does not match tables where type instance notifications are accepted on.");
    }
  }
  return containers;
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
/**
 * Parses the notifications from the database back end for local edge comments by using a regular
 * expression. If the regular expression matches the supplied {@link PGNotification} notification,
 * it is determined if the edge in question is loaded, and if a
 * {@link CommentNotificationContainer} is build with the data from the notification.
 *
 * @param notification The {@link PGNotification} from the PostgreSQL database server.
 * @param provider The {@link SQLProvider} which is used to communicate with the database.
 */
static CommentNotification processEdgeLocalCommentNotification(final PGNotification notification,
    final SQLProvider provider) {
  final Matcher matcher = EDGE_LOCAL_PATTERN.matcher(notification.getParameter());
  if (!matcher.find()) {
    return null;
  }

  final Integer edgeId = Integer.parseInt(matcher.group(3));
  final Integer commentId =
      matcher.group(4).equals("null") ? null : Integer.parseInt(matcher.group(4));

  final INaviEdge edge = EdgeCache.get(provider).getEdgeById(edgeId);
  if (edge == null) {
    return null;
  }

  final CommentOperation operation =
      commentId == null ? CommentOperation.DELETE : CommentOperation.APPEND;

  return new EdgeCommentNotificationContainer(edge, operation, CommentScope.LOCAL, commentId);
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
/**
 * Parses the notifications from the database back end for group node comments by using a regular
 * expression. If the regular expression matches the supplied {@link PGNotification} notification,
 * it is determined if the group node in the notification is currently loaded, and if a
 * {@link CommentNotificationContainer} with the gathered data from the notification is returned.
 *
 * @param notification The {@link PGNotification} from the PostgreSQL database server.
 * @param provider The {@link SQLProvider} which is used to communicate with the database.
 */
static CommentNotification processGroupNodeCommentNotification(final PGNotification notification,
    final SQLProvider provider) {
  final Matcher matcher = GROUP_NODE_PATTERN.matcher(notification.getParameter());
  if (!matcher.find()) {
    return null;
  }

  final Integer nodeId = Integer.parseInt(matcher.group(3));
  final Integer commentId =
      matcher.group(4).equals("null") ? null : Integer.parseInt(matcher.group(4));

  final INaviGroupNode groupNode = (INaviGroupNode) NodeCache.get(provider).getNodeById(nodeId);
  if (groupNode == null) {
    return null;
  }

  final CommentOperation operation =
      commentId == null ? CommentOperation.DELETE : CommentOperation.APPEND;

  return new GroupNodeCommentNotificationContainer(groupNode, operation, commentId);
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
/**
 * Parses the {@link PGNotification} notifications from the PostgreSQL database back end for text
 * node comments by using a regular expression. If the regular expression matches the supplied
 * {@link PGNotification} notification, it is determined if the text node in the notification is
 * currently loaded, and if a {@link CommentNotificationContainer} with the gathered data from the
 * notification is returned.
 *
 * @param notification The {@link PGNotification} from the PostgreSQL database server.
 * @param provider The {@link SQLProvider} which is used to communicate with the database.
 */
static CommentNotification processTextNodeCommentNotification(final PGNotification notification,
    final SQLProvider provider) {
  final Matcher matcher = TEXT_NODE_PATTERN.matcher(notification.getParameter());
  if (!matcher.find()) {
    return null;
  }

  final Integer nodeId = Integer.parseInt(matcher.group(3));
  final Integer commentId =
      matcher.group(4).equals("null") ? null : Integer.parseInt(matcher.group(4));

  final INaviTextNode node = (INaviTextNode) NodeCache.get(provider).getNodeById(nodeId);
  if (node == null) {
    return null;
  }

  final CommentOperation operation =
      commentId == null ? CommentOperation.DELETE : CommentOperation.APPEND;

  return new TextNodeCommentNotificationContainer(node, operation, commentId);
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
private FunctionNotificationContainer parseFunctionNotification(final PGNotification notification,
    final SQLProvider provider) {

  final Pattern pattern = Pattern.compile(functionNotificationPattern);
  final Matcher matcher = pattern.matcher(notification.getParameter());
  if (!matcher.find()) {
    throw new IllegalStateException("IE02739: compiled pattern: " + pattern.toString()
        + " did not match notification: " + notification.getParameter());
  }

  final String databaseOperation = matcher.group(2);
  final Integer moduleId = Integer.parseInt(matcher.group(3));
  final IAddress functionAddress = new CAddress(new BigInteger(matcher.group(4)));
  final INaviModule module = provider.findModule(moduleId);
  return new FunctionNotificationContainer(moduleId, module, functionAddress, databaseOperation);
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
@Override
public Collection<FunctionNotificationContainer> parse(
    final Collection<PGNotification> notifications, final SQLProvider provider) {

  Preconditions.checkNotNull(notifications, "IE02629: notifications argument can not be null");
  Preconditions.checkNotNull(provider, "IE02630: provider argument can not be null");

  final Collection<FunctionNotificationContainer> containers =
      new ArrayList<FunctionNotificationContainer>();

  for (final PGNotification notification : notifications) {
    if (notification.getParameter().startsWith(CTableNames.FUNCTIONS_TABLE)) {
      containers.add(parseFunctionNotification(notification, provider));
    } else {
      throw new IllegalStateException("IE02738: Table name supplied in notification: "
          + notification.getParameter()
          + " does not match tables where function notifications are accepted on.");
    }
  }
  return containers;
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
/**
 * Parser for a bn_views notification. The function uses the viewNotificationPattern to parse the
 * incoming {@link PGNotification} into a {@link ViewNotificationContainer}.
 *
 * @param notification The {@link PGNotification} to parse.
 * @param provider The {@link SQLProvider} to access the database.
 *
 * @return A {@link ViewNotificationContainer} with the parsed information.
 */
private ViewNotificationContainer parseViewNotification(
    final PGNotification notification, final SQLProvider provider) {

  final Matcher matcher = viewNotificationPattern.matcher(notification.getParameter());
  if (!matcher.find()) {
    throw new IllegalStateException("IE02742: compiled pattern: "
        + viewNotificationPattern.toString() + " did not match notification: "
        + notification.getParameter());
  }

  final Integer viewId = Integer.parseInt(matcher.group(3));
  final Optional<INaviView> view =
      Optional.fromNullable(ViewManager.get(provider).getView(viewId));
  final String databaseOperation = matcher.group(2);

  return new ViewNotificationContainer(viewId,
      view,
      Optional.<Integer>absent(),
      Optional.<INaviModule>absent(),
      Optional.<INaviProject>absent(),
      databaseOperation);
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
/**
 * Parser for a bn_module_views notification. The function uses the moduleViewNotificationPattern
 * to parse the incoming {@link PGNotification} into a {@link ViewNotificationContainer}.
 *
 * @param notification The {@link PGNotification} to parse.
 * @param provider The {@link SQLProvider} to access the database.
 *
 * @return A {@link ViewNotificationContainer} with the parsed information.
 */
private ViewNotificationContainer parseModuleViewNotification(
    final PGNotification notification, final SQLProvider provider) {

  final Matcher matcher = moduleViewNotificationPattern.matcher(notification.getParameter());
  if (!matcher.find()) {
    throw new IllegalStateException("IE02743: compiled pattern: "
        + moduleViewNotificationPattern.toString() + " did not match notification: "
        + notification.getParameter());
  }

  final Integer viewId = Integer.parseInt(matcher.group(3));
  final Optional<INaviView> view =
      Optional.fromNullable(ViewManager.get(provider).getView(viewId));
  final Optional<Integer> moduleId = Optional.fromNullable(Integer.parseInt(matcher.group(4)));
  final Optional<INaviModule> module = Optional.fromNullable(provider.findModule(moduleId.get()));
  final String databaseOperation = matcher.group(2);

  return new ViewNotificationContainer(viewId,
      view,
      moduleId,
      module,
      Optional.<INaviProject>absent(),
      databaseOperation);
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
/**
 * Parser for the bn_project_views notification. The function uses the
 * projectViewNotificationPattern to parse the incoming {@link PGNotification} into a
 * {@link ViewNotificationContainer}
 *
 * @param notification The {@link PGNotification} to parse.
 * @param provider The {@link SQLProvider} to access the database.
 */
private ViewNotificationContainer parseProjectViewNotification(
    final PGNotification notification, final SQLProvider provider) {

  final Matcher matcher = projectViewNotificationPattern.matcher(notification.getParameter());
  if (!matcher.find()) {
    throw new IllegalStateException("IE02744: compiled pattern: "
        + projectViewNotificationPattern.toString() + " did not match notification: "
        + notification.getParameter());
  }

  final Integer viewId = Integer.parseInt(matcher.group(3));
  final Optional<INaviView> view =
      Optional.fromNullable(ViewManager.get(provider).getView(viewId));
  final Optional<Integer> projectId = Optional.fromNullable(Integer.parseInt(matcher.group(4)));
  final Optional<INaviProject> project =
      Optional.fromNullable(provider.findProject(projectId.get()));
  final String databaseOperation = matcher.group(2);

  return new ViewNotificationContainer(viewId,
      view,
      projectId,
      Optional.<INaviModule>absent(),
      project,
      databaseOperation);
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
/**
 * The parser function for view notifications. The function checks for the database table the
 * notification originated from and dispatches the parsing appropriately to the right function.
 *
 * @param notifications The {@link Collection} of {@link PGNotification} to parse.
 * @param provider The {@link SQLProvider} to access the database with.
 * @return A {@link Collection} of {@link ViewNotificationContainer}.
 */
@Override
public Collection<ViewNotificationContainer> parse(
    final Collection<PGNotification> notifications, final SQLProvider provider) {
  Preconditions.checkNotNull(notifications, "IE02745: notifications argument can not be null");
  Preconditions.checkNotNull(provider, "IE02746: provider argument can not be null");

  final Collection<ViewNotificationContainer> containers = Lists.newArrayList();

  for (final PGNotification notification : notifications) {
    if (notification.getParameter().startsWith(CTableNames.VIEWS_TABLE)) {
      containers.add(parseViewNotification(notification, provider));
    } else if (notification.getParameter().startsWith(CTableNames.MODULE_VIEWS_TABLE)) {
      containers.add(parseModuleViewNotification(notification, provider));
    } else if (notification.getParameter().startsWith(CTableNames.PROJECT_VIEWS_TABLE)) {
      containers.add(parseProjectViewNotification(notification, provider));
    } else {
      throw new IllegalStateException("IE02747: Table name supplied in notification: "
          + notification.getParameter()
          + " does not match tables where view notifications are accepted on.");
    }
  }
  return containers;
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
@Test
public void testFunctionNodeCommentParsingCompleteGarbageInput() {
  new MockModule(provider, Lists.newArrayList(mockView), Lists.newArrayList(mockFunction));

  // Normal input "bn_function_nodes UPDATE 1 6666 4608 null"
  notifications.add(new MockPGNotification("comment_changes", "bn_function_nodes 1 1 1 1 1"));
  notifications.add(new MockPGNotification("comment_changes", "bn_function_nodes UPDATE"));
  notifications.add(new MockPGNotification("comment_changes", "bn_function_nodes UPDATE 1 "));
  notifications.add(new MockPGNotification("comment_changes", "bn_function_nodes"));
  notifications.add(new MockPGNotification("comment_changes",
      "bn_function_nodes 1 FOO FOO FOOO ooqwkepqwpoekpqowkep"
      + "oqw\\n\\\n\\\n\\\\\\n\\n\n\\n\\\n\\n\\\n\\c\\c\\c\\c"
      + "ckepokqwpekpqwokepoaksjeofijsoiefjosejfosjoefjsoisje" + "foisjefoisjeofijsoeifjsoeifj"));

  for (PGNotification notification : notifications) {
    assertNull(PostgreSQLCommentNotificationParser.processFunctionCommentNotification(
        notification, provider));
  }
}
 

import org.postgresql.PGNotification; //导入依赖的package包/类
@Test(groups = SMALL)
public void testGetNewEventsParsesCorrectlyTheNotificationsPaylodToCreateANewEvent() throws SQLException {
    mapper = new PsqlEventMapper();
    eventSource = new PsqlEventSource(connectionProvider, TABLE_NAME, mapper, WITH_AUTO_CONFIG);

    when(mockedPgConnection.getNotifications()).thenReturn(new PGNotification[] {
            new MyPGNotification(STREAM_NAME, VALID_PAYLOAD)
    });

    eventSource.connect();
    List<Event<Map<String, Object>>> newEvents = eventSource.getNewEvents();

    assertEquals(newEvents.size(), 1);

    Event<Map<String, Object>> event = newEvents.get(0);
    assertEquals(event.getEventType(), "INSERT");
    assertEquals(event.getEntityName(), "aTable");
    assertEquals(event.getNewEntity().get("a"), "b");
    assertEquals(event.getOldEntity(), new HashMap<String, String>());
}